import logging import re from pathlib import Path import cv2 import yaml from fastapi import HTTPException from markdown_it import MarkdownIt from mdit_py_plugins.anchors import anchors_plugin as section_header from pydantic import ValidationError from app.schema import ( DocumentCheckout, DocumentEmail, DocumentIndex, DocumentOrders, DocumentPage, DocumentProduct, DocumentProductIndex, DocumentSite, ) logger = logging.getLogger(__name__) def md(text: str) -> str: """Helper function to convert parsed markdown-formatted text to HTML.""" md = ( MarkdownIt("commonmark", {"breaks": True, "html": True}) .enable("table") .use(section_header) ) return md.render(text) def get_files_by_extension(folder: Path, extensions: list[str], ignore_list: list = [], relative_path: bool = False) -> list[Path]: """Getting list of files specified by extensions""" file_list = [] for extension in extensions: file_list.extend([f for f in list(folder.rglob(f'{extension}')) if f not in ignore_list]) if relative_path: for i in range(len(file_list)): file_list[i] = file_list[i].relative_to(folder) return file_list def set_default_product_meta_fields(block) -> dict[str, str]: """ Helper function to make sure all necessary fields expected by the document schema are present in dictionary. This prevents errors when in the text document some keys have been omitted (because unnecessary / empty), but are needed by the document scheme. """ if block['template'] == 'product': if 'code' not in block: block['code'] = '' elif 'product_id' not in block: block['product_id'] = '' elif 'price' not in block: block['price'] = 0 elif 'weight' not in block: block['weight'] = 0 elif 'customs_tariff_code' not in block: block['customs_tariff_code'] = '' elif 'origin_country_iso' not in block: block['origin_country_iso'] = '' elif 'category' not in block: block['category'] = '' if 'options' in block and len(block['options']) > 0: if 'size' not in block['options']: block['options']['size'] = None if 'style' not in block['options']: block['options']['style'] = None else: block['options'] = { 'size':None, 'style': None, } if 'inventory' in block and len(block['inventory']) > 0: for item in block['inventory']: if 'size' not in item: item['size'] = None if 'style' not in item: item['style'] = None if 'amount' not in item: item['amount'] = 0 else: block['inventory'] = [ { 'size': None, 'style': None, 'amount': 0 } ] return block def make_block_md(block: str) -> dict[str]: """Helper function to prepare a dictionary object for a block of type text. Block texts are free-form strings and not formatted with YAML, so we prepare the dictionary here. """ block.strip() if block != "": return {"value": block, "type": "text"} def make_block_image( repo_path: str, meta_path: str, block: dict[str, str] ) -> dict[str, int]: """Helper function to enrich the image block with a few more info about the file it points to. We make use of this when displaying the image in the frontend template and prepare the correct srcset list of sizes we can display for the given image. """ filepath = None if (meta_path != '.'): filepath = f"{repo_path}/{meta_path}/{block['url']}" else: filepath = f"{repo_path}/{block['url']}" img = cv2.imread(filepath) if img is not None and img.any(): if 'caption' not in block: block['caption'] = "" block["info"] = { "width": round(img.shape[1]), "height": round(img.shape[0]), } return block else: return { "type": "image", "value": "", "url": "", "caption": "", "info": { "width": 0, "height": 0 }, } def parse_file( repo_path: str, input_path: str | Path, block_types: list[str] ) -> ( DocumentSite | DocumentIndex | DocumentPage | DocumentProductIndex | DocumentProduct | DocumentCheckout | DocumentOrders | DocumentEmail ): """Parse file at given filepath: return a dictionary with meta dictionary and list of blocks. """ logger.error(f"PARSE_FILE DEBUG: Parsing file: {input_path}") logger.error(f"PARSE_FILE DEBUG: Absolute path: {Path(input_path).absolute()}") logger.error(f"PARSE_FILE DEBUG: File exists: {Path(input_path).exists()}") # Read first few lines for debugging try: with open(input_path, 'r') as f: first_lines = [next(f) for _ in range(5)] logger.error(f"PARSE_FILE DEBUG: First 5 lines of file:") for i, line in enumerate(first_lines): logger.error(f" Line {i}: {line.strip()}") except Exception as e: logger.error(f"PARSE_FILE DEBUG: Could not read file: {e}") document = { "meta": { "type": "", "title": "", "template": "", "path": "", }, "blocks": [], } try: p = Path(input_path) except KeyError: document["meta"]["path"] = "" data = input_path.read_text() block_marker = "\n---\n" blocks = data.split(block_marker) blocks = [re.sub(r"^---\n", "", block) for block in blocks] # set meta_path to default value meta_path = document["meta"]["path"] for block in blocks: # we check if current block is a potential # yaml block, then check if any other block # has a field type in the ones "allowed". # in case the block.type has no matches # or it fails the yaml parser, we take it # as a markdown block try: block_yaml = yaml.safe_load(block) if block_yaml is not None: if type(block_yaml) is dict: # check if "valid" YAML dict is actually # a long string with a colon in the middle; # somehow this trips up the YAML parser and # we need to take care of it ourselves for k in block_yaml.keys(): if len(k.split()) > 1 and ":" in k: block_txt = make_block_md(block) document["blocks"].append(block_txt) if "type" in block_yaml: if block_yaml["type"] == "meta": document["meta"] = set_default_product_meta_fields(block_yaml) for k, v in block_yaml.items(): document["meta"][k] = v # set meta-path if "template" in block_yaml: doc_path = Path(p).relative_to(repo_path).parent document["meta"]["path"] = str(doc_path) meta_path = document["meta"]["path"] elif ( "type" in block_yaml and block_yaml["type"].lower() in block_types ): if block_yaml["type"] == "image": make_block_image(repo_path, meta_path, block_yaml) document["blocks"].append(block_yaml) else: block_md = make_block_md(block) document["blocks"].append(block_md) except yaml.YAMLError: block_md = make_block_md(block) document["blocks"].append(block_md) logger.error("=== DEBUG PARSED DOCUMENT ===") logger.error(f"File being parsed: {input_path}") logger.error(f"Document meta keys: {list(document['meta'].keys())}") logger.error(f"Document meta values: {document['meta']}") logger.error(f"Template found: '{document['meta'].get('template', 'NOT FOUND')}'") logger.error(f"Number of content blocks: {len(document['blocks'])}") # Check if template is missing if not document['meta'].get('template'): logger.error("CRITICAL: No template found in document!") if blocks and len(blocks) > 0: logger.error(f"First block preview: {blocks[0][:500]}") try: if document["meta"]["template"] == "site": return DocumentSite(**document) elif document["meta"]["template"] == "index": return DocumentIndex(**document) elif document["meta"]["template"] == "page": return DocumentPage(**document) elif document["meta"]["template"] == "product-index": return DocumentProductIndex(**document) elif document["meta"]["template"] == "product": return DocumentProduct(**document) elif document["meta"]["template"] == "checkout": return DocumentCheckout(**document) elif document["meta"]["template"] == "orders": return DocumentOrders(**document) elif document["meta"]["template"] == "email": return DocumentEmail(**document) except ValidationError as e: logger.error("pydantic validation errors =>") for e in e.errors(): logger.error(f"error => {e}") # Add more context about what failed logger.error(f"Validation failed for template: '{document['meta'].get('template', 'NO TEMPLATE')}'") logger.error(f"Document structure had keys: {list(document.keys())}") if 'meta' in document: logger.error(f"Meta had keys: {list(document['meta'].keys())}") raise FileNotFoundError def read_dir( repo_path: str, document_match: list[str], document_exclude: list[str], block_types: list[str], exclude_doc: str | None = None, tree: str | None = None, ): """Run get_files_by_extension with appropriate pattern matching to return list""" d = Path(repo_path).expanduser() pattern_matching = [] [pattern_matching.append(f"*{ext}") for ext in document_match] ignore_list = [] for file in document_exclude: ignore_list.append(Path(f"{repo_path}/{file}")) paths = get_files_by_extension(d, pattern_matching, ignore_list) paths = [p for p in paths if tree in str(p)] docs = [] for p in paths: doc = parse_file(d, p, block_types) if doc.meta.path != exclude_doc: docs.append(doc) return docs def read_file( repo_path: str, filename: str, file_match: list[str], block_types: list[str], tree: str | None = None, ): """Helper function to read file from disk. Check if filepath's extension is allowed (by settings.toml) and filepath is tracked by git, then read its content. """ filepath = None ext = file_match[0] if tree: p = f"{Path(repo_path)}/{tree}/{filename}" else: p = f"{Path(repo_path)}/{filename}" logger.error(f"READ_FILE DEBUG: Looking for file at path: {p}") logger.error(f"READ_FILE DEBUG: tree={tree}, filename={filename}, ext={ext}") if Path(p).is_dir(): file_path = f"{p}/index{ext}" logger.error(f"READ_FILE DEBUG: Path is directory, trying: {file_path}") if Path(file_path).is_file(): filepath = file_path logger.error(f"READ_FILE DEBUG: Found directory index: {filepath}") else: logger.error(f"READ_FILE DEBUG: Directory index not found: {file_path}") raise FileNotFoundError elif Path(f"{p}{ext}").is_file(): filepath = f"{p}{ext}" logger.error(f"READ_FILE DEBUG: Found file directly: {filepath}") if filepath: logger.error(f"READ_FILE DEBUG: Reading file: {filepath}") logger.error(f"READ_FILE DEBUG: File exists: {Path(filepath).exists()}") logger.error(f"READ_FILE DEBUG: File size: {Path(filepath).stat().st_size if Path(filepath).exists() else 'N/A'}") # get file path relative to the content repo # eg => / filename = str(Path(filepath).relative_to(repo_path)) doc = parse_file(repo_path, Path(filepath), block_types) return doc else: logger.error(f"READ_FILE DEBUG: No file found at: {p}{ext}") raise HTTPException(status_code=404, detail="Nothing was found here.")