diff --git a/requirements.txt b/requirements.txt index b212654..3299817 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,5 +14,6 @@ statsmodels>=0.14 plotly>=5.20 python-dotenv>=1.0 httpx>=0.27 +python-multipart>=0.0.9 pytest>=8 httpx2>=0.3 diff --git a/src/dashboard/app.py b/src/dashboard/app.py index 0213904..199cfd7 100644 --- a/src/dashboard/app.py +++ b/src/dashboard/app.py @@ -13,7 +13,7 @@ from typing import Any import plotly.graph_objects as go import plotly.io as pio -from fastapi import FastAPI, Query +from fastapi import FastAPI, Query, UploadFile, File, Form from fastapi.responses import HTMLResponse, JSONResponse from .. import db, config @@ -25,6 +25,7 @@ from ..analytics import ( ) from ..models import RawSnapshot from ..pipeline import orchestrator as orch +from ..pipeline import upload as upload_mod log = logging.getLogger(__name__) THERE = Path(__file__).parent @@ -488,6 +489,27 @@ def sources() -> HTMLResponse: return _render("sources", SOURCES_HTML="".join(rows)) +@app.post("/api/upload") +async def api_upload(file: UploadFile = File(...), + source_id: str = Form(...), + source_name: str = Form(""), + source_url: str = Form(""), + as_of: str = Form("")) -> JSONResponse: + """Accept a CSV/HTML file, parse prices, run through the pipeline, upsert. + Returns a summary (loaded / quarantined / skipped).""" + content = await file.read() + try: + res = upload_mod.upload(file.filename or "", content, + source_id=source_id, source_name=source_name, + source_url=source_url, as_of=as_of) + except Exception as e: + log.exception("upload failed") + return JSONResponse({"ok": False, "error": f"{type(e).__name__}: {e}"}, status_code=400) + res["ok"] = True + res["filename"] = file.filename + return JSONResponse(res) + + # ---------- API ---------- @app.get("/api/series") def api_series(product: str, country: str = "", region: str = "", diff --git a/src/dashboard/templates/sources.html b/src/dashboard/templates/sources.html index c72e33d..42abc77 100644 --- a/src/dashboard/templates/sources.html +++ b/src/dashboard/templates/sources.html @@ -13,3 +13,45 @@ {{ SOURCES_HTML }} + +
+

Ручная загрузка файла (Arbuz.kz, Sharyn и аналоги без API)

+

+ CSV со столбцами product, price, unit, currency, region, as_of (русские/английские названия + столбцов узнаются) или HTML-таблица (имя в 1-й колонке, цена во 2-й). Запись проходит через + тот же пайплайн: нормализация → категоризация → валидация. Низкая уверенность — в очередь на проверку, + не в аналитику. +

+
+ + + + + +
+ + +
+
+
+ + diff --git a/src/pipeline/upload.py b/src/pipeline/upload.py new file mode 100644 index 0000000..694c03e --- /dev/null +++ b/src/pipeline/upload.py @@ -0,0 +1,224 @@ +"""Manual CSV/HTML upload for sources without a public API (Arbuz.kz, Sharyn). + +Parses a user-supplied file into price records and runs them through the same +normalize → categorize → validate pipeline as the fetcher, then upserts. +Records that fail categorization land in the review queue (not in analytics). + +CSV format (header row required). Recognised column names (case-insensitive, +synonyms accepted): + - product name : product, name, товар, наименование + - price value : price, value, цена, стоим + - unit : unit, ед, единица + - currency : currency, code, вал + - region : region, area, регион, область + - as_of : as_of, date, дата + - source url : source_url, url + - market : market, рынок +Rows may use a per-region override for region. Missing as_of → today. +""" +from __future__ import annotations +import csv +import io +import re +from datetime import date, datetime +from typing import Optional + +from .. import config, db +from ..models import ExtractedPrice, SourceMeta +from .normalize import normalize_price +from .categorize import categorize_record +from .validate import validate_price + +# Column-name synonym map → canonical key +_COL_MAP = { + "product": "product", "name": "product", "product_name": "product", + "товар": "product", "наименование": "product", "наименованиe": "product", + "price": "price", "value": "price", "цена": "price", "стоим": "price", + "ед": "unit", "единица": "unit", "unit": "unit", + "currency": "currency", "code": "currency", + "вал": "currency", "currency_code": "currency", + "region": "region", "area": "region", "регион": "region", "область": "region", + "as_of": "as_of", "date": "as_of", "дата": "as_of", "asof": "as_of", + "source_url": "source_url", "url": "source_url", "ссылка": "source_url", + "market": "market", "рынок": "market", +} + +_NUMBER_RE = re.compile(r"(\d+(?:[.,]\d+)?)") + + +def _canonical_header(h: str) -> str: + return _COL_MAP.get((h or "").strip().lower(), (h or "").strip().lower()) + + +def _parse_num(s: str) -> Optional[float]: + if s is None: + return None + s = str(s).strip() + if not s: + return None + m = _NUMBER_RE.search(s.replace("\u00a0", " ").replace(" ", ".")) + if not m: + return None + return float(m.group(1)) + + +def _parse_date(s: str, default: date) -> date: + s = (s or "").strip() + if not s: + return default + for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y", "%d.%m.%y", "%m/%d/%Y"): + try: + return datetime.strptime(s, fmt).date() + except ValueError: + continue + return default + + +def _country_from_region(region: str) -> str: + return region.split("-")[0] if "-" in region else region + + +def _build_source(source_id: str, source_name: str, source_url: str) -> SourceMeta: + return SourceMeta( + id=source_id, name=source_name or source_id, url=source_url or "manual-upload", + tier=3, countries=("KZ", "TJ", "UZ", "RU"), legal_status="approved", + adapter="manual_csv", note="manual CSV/HTML upload", + ) + + +def _ingest_record(source: SourceMeta, rec: dict, raw_snapshot_id: Optional[str], + source_url: str, as_of_default: date) -> dict: + """Run one record through normalize → categorize → validate → upsert.""" + region = rec.get("region") or "KZ" + ep = ExtractedPrice( + raw_snapshot_id=raw_snapshot_id or "upload", + source_id=source.id, region=region, market=rec.get("market"), + product_name=rec.get("product") or "unknown", + raw_value=str(rec.get("price")), value=_parse_num(str(rec.get("price"))) or 0.0, + unit=rec.get("unit") or "kg", currency=(rec.get("currency") or "KZT").upper(), + price_type=rec.get("price_type") or "retail", + as_of=_parse_date(rec.get("as_of"), as_of_default), + fetched_at=datetime.utcnow(), + source_url=source_url, + ) + nrec = normalize_price(ep, as_of=ep.as_of) + nrec["_product_name"] = rec.get("product") + nrec["source_id"] = source.id + categorize_record(nrec) + nrec, _ = validate_price(nrec) + if raw_snapshot_id: + nrec["raw_snapshot_id"] = raw_snapshot_id + db.upsert_price(nrec) + return nrec + + +def parse_csv(text: str, source_id: str, source_name: str = "", + source_url: str = "", as_of_str: str = "") -> dict: + """Parse CSV text into prices. Returns a summary dict.""" + as_of_default = _parse_date(as_of_str, date.today()) + rows = _rows_from_csv(text) + source = _build_source(source_id, source_name, source_url) + raw_snapshot_id = None + if text: + snap = db.save_raw(source.id, source_url or "manual-upload", text.encode("utf-8"), "text/csv") + raw_snapshot_id = snap.id + loaded = quarantined = skipped = 0 + for r in rows: + product = (r.get("product") or "").strip() + price = _parse_num(r.get("price")) + if not product or price is None or price <= 0: + skipped += 1 + continue + rec = { + "product": product, "price": price, + "unit": r.get("unit") or "kg", "currency": r.get("currency") or "KZT", + "region": r.get("region") or "KZ", "as_of": r.get("as_of"), + "market": r.get("market"), + } + out = _ingest_record(source, rec, raw_snapshot_id, source_url, as_of_default) + if out.get("quarantined"): + quarantined += 1 + else: + loaded += 1 + return {"loaded": loaded, "quarantined": quarantined, "skipped": skipped, + "total": len(rows), "source_id": source.id} + + +def _rows_from_csv(text: str) -> list["dict"]: + text = text.lstrip("\ufeff") + reader = csv.reader(io.StringIO(text)) + rows = [r for r in reader if any((c or "").strip() for c in r)] + if not rows: + return [] + header = [_canonical_header(h) for h in rows[0]] + out = [] + for row in rows[1:]: + d = {} + for i, val in enumerate(row): + if i < len(header): + d[header[i]] = (val or "").strip() + out.append(d) + return out + + +_HTML_NUM = re.compile(r"(\d[\d.,\u00a0]*\d|\d)") + + +def parse_html(html_text: str, source_id: str, source_name: str = "", + source_url: str = "", as_of_str: str = "") -> dict: + """Parse an HTML price table: rows with a name cell + a numeric price cell. + Assumes the first with >2 rows. Returns summary dict.""" + from bs4 import BeautifulSoup + soup = BeautifulSoup(html_text, "lxml") + table = soup.find("table") + if table is None: + return {"loaded": 0, "quarantined": 0, "skipped": 0, "total": 0, + "source_id": source_id, "error": "no
found in HTML"} + as_of_default = _parse_date(as_of_str, date.today()) + source = _build_source(source_id, source_name, source_url) + snap = db.save_raw(source.id, source_url or "manual-upload", + html_text.encode("utf-8"), "text/html") + loaded = quarantined = skipped = 0 + body = table.find("tbody") or table + for tr in body.find_all("tr"): + cells = [td for td in tr.find_all(["td", "th"])] + if len(cells) < 2: + continue + name = "".join(cells[0].get_text(" ", strip=True).split()) + # price = first numeric-looking cell after the name + price = None + for c in cells[1:]: + t = "".join(c.get_text(" ", strip=True).split()) + m = _HTML_NUM.search(t.replace("\u00a0", "")) + if m: + price = _parse_num(t) + break + if not name or price is None or price <= 0: + skipped += 1 + continue + rec = {"product": name, "price": price, "unit": "kg", "currency": "KZT", + "region": "KZ", "as_of": None, "market": source_name or None} + out = _ingest_record(source, rec, snap.id, source_url, as_of_default) + if out.get("quarantined"): + quarantined += 1 + else: + loaded += 1 + return {"loaded": loaded, "quarantined": quarantined, "skipped": skipped, + "total": 0, "source_id": source.id} + + +def upload(filename: str, content: bytes, source_id: str, source_name: str = "", + source_url: str = "", as_of: str = "") -> dict: + """Dispatch by file type. Returns summary dict.""" + name = filename or "" + low = name.lower() + text = content.decode("utf-8", errors="replace") + if low.endswith(".csv") or ".csv" in low or low.endswith(".txt"): + return parse_csv(text, source_id=source_id, source_name=source_name, + source_url=source_url, as_of_str=as_of) + if low.endswith((".html", ".htm")) or "= 3 + + def test_low_confidence_goes_to_review(self): + from src.pipeline.upload import parse_csv + from src import db + text = "product,price\nЗерно зерновое суперспецифическое,100\n" + res = parse_csv(text, source_id="up_rev", as_of_str="2026-09-24") + # Unmatched product → quarantined (not loaded) + assert res["quarantined"] >= 1 + reviews = db.pending_reviews(limit=50) + assert any("Зерно зерновое суперспецифическое" in (r.get("payload") or {}).get("product_name", "") + for r in reviews) + + class TestDigest: def test_build_digest(self): from src.digest import build_digest