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. """ 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) 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}") 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}" if Path(p).is_dir(): file_path = f"{p}/index{ext}" if Path(file_path).is_file(): filepath = file_path else: raise FileNotFoundError elif Path(f"{p}{ext}").is_file(): filepath = f"{p}{ext}" if filepath: # 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: raise HTTPException(status_code=404, detail="Nothing was found here.")