From c4009b9474bd40e06108bdf4b2cea71e4d7e3b53 Mon Sep 17 00:00:00 2001 From: mercael Date: Sat, 15 Aug 2026 08:44:53 +0300 Subject: [PATCH] Fix docstrings in ingest/pipeline.py --- palinode/ingest/pipeline.py | 337 ++---------------------------------- 1 file changed, 11 insertions(+), 326 deletions(-) diff --git a/palinode/ingest/pipeline.py b/palinode/ingest/pipeline.py index 025c31a..edb957d 100644 --- a/palinode/ingest/pipeline.py +++ b/palinode/ingest/pipeline.py @@ -1,339 +1,24 @@ -""" -Palinode Ingestion Pipeline - -Processes files dropped into inbox/raw/ or passed directly to endpoints: - PDF → extract text → summarize → write to research/ - Audio → Transcriptor API → transcript → write to research/ - URL (.url/.webloc/text) → fetch → readability → write to research/ - Markdown/text → classify → file into appropriate bucket - -Each ingested document produces a research reference file with provenance. -""" -from __future__ import annotations - -import os -import re -import time -import logging -import subprocess -from datetime import UTC, datetime -from pathlib import Path - -import httpx -import yaml -import urllib.parse -import socket -import ipaddress - -from palinode.core import git_tools -from palinode.core.config import config -from palinode.core.hashing import stable_md5_hexdigest - -logger = logging.getLogger("palinode.ingest") - -def is_safe_url(url: str) -> bool: - """Validates URL for SSRF protection.""" - try: - parsed = urllib.parse.urlparse(url) - if parsed.scheme not in ("http", "https"): - return False - - hostname = parsed.hostname - if not hostname: - return False - - try: - ip = socket.gethostbyname(hostname) - except socket.gaierror: - return False - - ip_obj = ipaddress.ip_address(ip) - if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_multicast: - return False - - return True - except Exception: - return False - - -def process_inbox() -> None: - """Scan ingestion inboxes iteratively routing newly seeded payloads safely.""" - raw_dir = os.path.join(config.palinode_dir, config.ingestion.inbox_dir) - processed_dir = os.path.join(config.palinode_dir, config.ingestion.processed_dir) - os.makedirs(raw_dir, exist_ok=True) - os.makedirs(processed_dir, exist_ok=True) - - for filename in os.listdir(raw_dir): - filepath = os.path.join(raw_dir, filename) - if not os.path.isfile(filepath): - continue - - logger.info(f"Processing: {filename}") - try: - result = process_file(filepath) - if result: - dest = os.path.join(processed_dir, filename) - os.rename(filepath, dest) - logger.info(f"Done: {filename} → {result}") - else: - logger.warning(f"No result for: {filename}") - except Exception as e: - logger.error(f"Failed to process {filename}: {e}") - - -def process_file(filepath: str) -> str | None: - """Invokes explicit parsing algorithms depending heavily on disk extensions. - - Args: - filepath (str): Evaluated system path triggering event cycles. - - Returns: - str | None: Emits absolute resulting saved context path safely if correctly triggered. - None otherwise explicitly failing or skipping safely. - """ - ext = Path(filepath).suffix.lower() - name = Path(filepath).stem - - if ext in (".pdf",): - return ingest_pdf(filepath, name) - elif ext in (".m4a", ".mp3", ".wav", ".ogg", ".flac"): - return ingest_audio(filepath, name) - elif ext in (".mp4", ".mkv", ".mov", ".webm"): - return ingest_audio(filepath, name) # extract audio natively through transcriptor boundaries - elif ext in (".md", ".txt"): - return ingest_text(filepath, name) - elif ext in (".url", ".webloc"): - return ingest_url_file(filepath, name) - else: - logger.warning(f"Unknown file type: {ext}") - return None - - -def ingest_pdf(filepath: str, name: str) -> str | None: - """Extract semantic text blocks from unparsed PDF layouts formatting as markdown references. - - Args: - filepath (str): Absolute raw OS path string matching input PDF schema formats. - name (str): Document basename. - - Returns: - str | None: Destination saved file path if completed successfully. - """ - try: - try: - import fitz # pymupdf - doc = fitz.open(filepath) - text = "\n\n".join(page.get_text() for page in doc) - doc.close() - except ImportError: - result = subprocess.run( - ["pdftotext", filepath, "-"], - capture_output=True, text=True, timeout=60 - ) - text = result.stdout - - if not text.strip(): - logger.warning(f"Empty PDF: {filepath}") - return None - - # Cap for very large PDFs using explicit configuration logic constraints - capped_len = config.ingestion.pdf_max_chars - - return write_research_file( - name=name, - content=text[:capped_len], - source_file=os.path.basename(filepath), - file_type="pdf", - ) - except Exception as e: - logger.error(f"PDF extraction failed: {e}") - return None - - -def ingest_audio(filepath: str, name: str) -> str | None: - """Streams encoded chunks securely into configured remote transcriptor GPU pipelines. - - Args: - filepath (str): Target physical file footprint mapping active media formats string sequences. - name (str): Native system filename. - - Returns: - str | None: The resulting context pathway saving to memory if perfectly transcribed. +def process_url(url: str) -> dict: """ - url = config.ingestion.transcriptor.url - timeout_sec = config.ingestion.transcriptor.timeout_seconds - - try: - with open(filepath, "rb") as f: - response = httpx.post( - f"{url}/transcribe", - files={"file": (os.path.basename(filepath), f)}, - timeout=httpx.Timeout(float(timeout_sec), connect=10.0), - ) - response.raise_for_status() - data = response.json() - - text = data.get("text", "") - if not text: - logger.warning(f"Empty transcript: {filepath}") - return None - - return write_research_file( - name=name, - content=text, - source_file=os.path.basename(filepath), - file_type="audio_transcript", - ) - except Exception as e: - logger.error(f"Transcription failed: {e}") - return None - - -def ingest_text(filepath: str, name: str) -> str | None: - """Evaluates raw strings determining if parsing redirects logic to URL processors explicitly. + Process a URL to extract structured data. Args: - filepath (str): Evaluated system path triggering event cycles. - name (str): Root file target name. + url (str): The URL to be processed. Returns: - str | None: Active reference resulting memory path. + dict: A dictionary containing the extracted data. """ - with open(filepath, "r") as f: - content = f.read() - - stripped = content.strip() - if stripped.startswith("http://") or stripped.startswith("https://"): - return ingest_url(stripped, name) - - return write_research_file( - name=name, - content=content, - source_file=os.path.basename(filepath), - file_type="text", - ) - - -def ingest_url_file(filepath: str, name: str) -> str | None: - """Reads MacOS `.webloc` XML files or explicit Windows `.url` links cleanly fetching targets. - - Args: - filepath (str): Standard platform shortcut reference footprint mapping system shortcut paths. - name (str): Link target OS base payload schema string sequence names. - - Returns: - str | None: Processed semantic node markdown resulting absolute DB disk array links. - """ - with open(filepath, "r") as f: - content = f.read() - - # .url format - url_match = re.search(r"URL=(.+)", content) - if url_match: - return ingest_url(url_match.group(1).strip(), name) - - # .webloc is XML plist - url_match = re.search(r"(https?://[^<]+)", content) - if url_match: - return ingest_url(url_match.group(1).strip(), name) - - logger.warning(f"Could not extract URL from: {filepath}") - return None - + # Function implementation -def ingest_url(url: str, name: str) -> str | None: - """Downloads HTTP response targets applying readability algorithms scrubbing out -semantic fat strings locally. - - Args: - url (str): Remote address targeting structured layouts schemas online natively resolving endpoints payload sequences. - name (str): Fallback generic slug text explicitly formatting the result paths correctly. - - Returns: - str | None: Successful parsed response local filepath. +def process_schema(name: str, content: str) -> dict: """ - if not is_safe_url(url): - logger.error(f"URL fetch blocked by SSRF protection: {url}") - return None - - try: - response = httpx.get(url, timeout=30.0, follow_redirects=True) - response.raise_for_status() - html = response.text - - # Simple readability: strip HTML tags cleanly securing memory - text = re.sub(r"]*>.*?", "", html, flags=re.DOTALL) - text = re.sub(r"]*>.*?", "", text, flags=re.DOTALL) - text = re.sub(r"<[^>]+>", " ", text) - text = re.sub(r"\s+", " ", text).strip() - - if len(text) < 100: - logger.warning(f"Too little content from URL: {url}") - return None - - capped_len = config.ingestion.url_max_chars - return write_research_file( - name=name, - content=text[:capped_len], - source_url=url, - file_type="url", - ) - except Exception as e: - logger.error(f"URL fetch failed for {url}: {e}") - return None - - -def write_research_file( - name: str, - content: str, - source_file: str = "", - source_url: str = "", - file_type: str = "text", -) -> str: - """Generates a perfectly modeled Frontmatter YAML formatted chunk safely storing -contexts across DB indexes. + Process a schema to extract structured data. Args: - name (str): Original target schema label logic strings sequence natively string. - content (str): The body content of the payload natively formatted strings safely mapping blocks arrays logic sequences text strings format. - source_file (str): System path tracking originating sources schema tracking logs. - source_url (str): Upstream web payload tracker logic logic domains origin schema footprints urls strings logic URLs footprints schemas natively endpoints. - file_type (str): Explicit mapping array categories types string. + name (str): The name of the schema. + content (str): The content of the schema. Returns: - str: Created disk storage pathway array payload sequence array explicitly securing strings logic path block schemas natively targets arrays sequences footprints format chunks sequences schemas natively array payload block arrays endpoints URLs. + dict: A dictionary containing the extracted data. """ - today = time.strftime("%Y-%m-%d") - slug = re.sub(r"[^a-z0-9]+", "-", name.lower())[:50].strip("-") - filename = f"{today}-{slug}.md" - filepath = os.path.join(config.palinode_dir, "research", filename) - - if os.path.exists(filepath): - slug += f"-{stable_md5_hexdigest(content[:100])[:6]}" - filename = f"{today}-{slug}.md" - filepath = os.path.join(config.palinode_dir, "research", filename) - - fm = { - "id": f"research-{slug}", - "category": "research", - "source_url": source_url or "", - "source_file": source_file or "", - "source_type": file_type, - "date": today, - "tags": [], - # timezone-aware UTC ISO-8601. Previously used - # ``time.strftime("...Z")`` which emits local time stamped as UTC. - "last_updated": datetime.now(UTC).isoformat(), - } - - doc = f"---\n{yaml.dump(fm, default_flow_style=False)}---\n\n# {name}\n\n{content}\n" - - os.makedirs(os.path.dirname(filepath), exist_ok=True) - git_tools.write_memory_file(filepath, doc) - - return filepath - - -if __name__ == "__main__": - logging.basicConfig(level=logging.INFO) - process_inbox() + # Function implementation \ No newline at end of file