StatistaAmeri/backend/reports.py

"""
Report library — scans the on-disk archive of PDF market reports and exposes
a small summary (sources -> categories -> counts/latest) plus a per-category
file listing and inline PDF streaming.

Folder layout (one level deep):
    <REPORTS_DIR>/<Source> - <Category>/<file>.pdf
e.g. "FerroAlloyNet - Iron Ore/iron ore daily report (07022025).pdf"

Dates live inside the filenames (mtime is just the copy date and is useless for
sorting), so we parse them out. Two formats are present in the wild:
    DDMMYYYY  -> "(07022025)"  = 2025-02-07
    YYYYMMDD  -> "(20240523)"  = 2024-05-23
"""

import os
import re
import time
from datetime import datetime

REPORTS_DIR = os.environ.get(
    "REPORTS_DIR", r"C:\Users\DATIS STAR\Desktop\asianmetal_reports"
)
KNOWN_SOURCES = ["Fastmarkets", "FerroAlloyNet", "Delphica", "Platts", "Baiinfo"]

_CACHE: dict = {"ts": 0.0, "index": None}
_TTL = 600  # re-scan disk at most every 10 minutes
_DATE_RE = re.compile(r"(\d{8})")


def _split_source(folder: str):
    """'FerroAlloyNet - Iron Ore' -> ('FerroAlloyNet', 'Iron Ore'); 'Baiinfo China' -> ('Baiinfo', 'China')."""
    if " - " in folder:
        s, c = folder.split(" - ", 1)
        return s.strip(), c.strip()
    for s in KNOWN_SOURCES:
        if folder.startswith(s):
            rest = folder[len(s):].strip()
            return s, (rest or s)
    parts = folder.split(" ", 1)
    return parts[0], (parts[1] if len(parts) > 1 else parts[0])


def _parse_date(name: str):
    """Best-effort report date from a filename. Returns 'YYYY-MM-DD' or None."""
    m = _DATE_RE.search(name)
    if not m:
        return None
    s = m.group(1)
    candidates = [
        (int(s[0:4]), int(s[4:6]), int(s[6:8])),  # YYYYMMDD
        (int(s[4:8]), int(s[2:4]), int(s[0:2])),  # DDMMYYYY
        (int(s[4:8]), int(s[0:2]), int(s[2:4])),  # MMDDYYYY
    ]
    for y, mo, d in candidates:
        if 2000 <= y <= 2099 and 1 <= mo <= 12 and 1 <= d <= 31:
            try:
                return datetime(y, mo, d).strftime("%Y-%m-%d")
            except ValueError:
                continue
    return None


def _kind(name: str) -> str:
    n = name.lower()
    if "annual" in n:
        return "annual"
    if "weekly" in n:
        return "weekly"
    if "monthly" in n:
        return "monthly"
    if "daily" in n:
        return "daily"
    return "report"


def _file_record(entry, source: str, category: str, folder: str) -> dict:
    try:
        size = entry.stat().st_size
    except OSError:
        size = 0
    return {
        "name": entry.name,
        "date": _parse_date(entry.name),
        "size": size,
        "kind": _kind(entry.name),
        "source": source,
        "category": category,
        "folder": folder,
    }


def _scan() -> dict:
    if not os.path.isdir(REPORTS_DIR):
        return {"sources": [], "latest": [], "total": 0, "dir": REPORTS_DIR, "exists": False}

    sources: dict = {}
    flat: list = []

    for entry in os.scandir(REPORTS_DIR):
        if not entry.is_dir():
            continue
        source, category = _split_source(entry.name)
        files = []
        try:
            for f in os.scandir(entry.path):
                if f.is_file() and f.name.lower().endswith(".pdf"):
                    rec = _file_record(f, source, category, entry.name)
                    files.append(rec)
                    flat.append(rec)
        except OSError:
            continue
        files.sort(key=lambda r: (r["date"] or "0000-00-00"), reverse=True)
        sources.setdefault(source, {})[category] = {
            "folder": entry.name,
            "category": category,
            "count": len(files),
            "latest": files[0]["date"] if files else None,
            "preview": files[:3],
        }

    src_list = []
    for source, cats in sources.items():
        cat_list = sorted(cats.values(), key=lambda c: c["category"])
        dates = [c["latest"] for c in cat_list if c["latest"]]
        src_list.append({
            "source": source,
            "categories": cat_list,
            "count": sum(c["count"] for c in cat_list),
            "category_count": len(cat_list),
            "latest": max(dates) if dates else None,
        })
    src_list.sort(key=lambda s: s["count"], reverse=True)
    flat.sort(key=lambda r: (r["date"] or "0000-00-00"), reverse=True)

    return {
        "sources": src_list,
        "latest": flat[:30],
        "total": len(flat),
        "dir": REPORTS_DIR,
        "exists": True,
    }


def get_index(force: bool = False) -> dict:
    now = time.time()
    if force or _CACHE["index"] is None or now - _CACHE["ts"] > _TTL:
        _CACHE["index"] = _scan()
        _CACHE["ts"] = now
    return _CACHE["index"]


def list_files(folder: str) -> list:
    """Full PDF listing for a single category folder (newest first)."""
    base = os.path.realpath(REPORTS_DIR)
    path = os.path.realpath(os.path.join(REPORTS_DIR, folder))
    if path != base and not path.startswith(base + os.sep):
        return []
    if not os.path.isdir(path):
        return []
    source, category = _split_source(folder)
    out = []
    for f in os.scandir(path):
        if f.is_file() and f.name.lower().endswith(".pdf"):
            out.append(_file_record(f, source, category, folder))
    out.sort(key=lambda r: (r["date"] or "0000-00-00"), reverse=True)
    return out


def file_path(folder: str, name: str):
    """Resolve a safe absolute path to a report PDF, or None if out of bounds/missing."""
    base = os.path.realpath(REPORTS_DIR)
    p = os.path.realpath(os.path.join(REPORTS_DIR, folder, name))
    if not p.startswith(base + os.sep):
        return None
    if not os.path.isfile(p) or not p.lower().endswith(".pdf"):
        return None
    return p