feat: manual CSV/HTML upload for sources without public API (Arbuz/Sharyn)

- src/pipeline/upload.py: parse_csv/parse_html → normalize→categorize→validate→upsert
- POST /api/upload (multipart) + form on /sources
- low-confidence records → review queue, not analytics
- +3 tests (16 passed, 1 skipped)

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

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 = "",

src/dashboard/templates/sources.html

@@ -13,3 +13,45 @@
     <tbody>{{ SOURCES_HTML }}</tbody>
   </table>
 </div>
+
+<div class="panel">
+  <h3>Ручная загрузка файла (Arbuz.kz, Sharyn и аналоги без API)</h3>
+  <p class="muted" style="font-size:13px">
+    CSV со столбцами <code>product, price, unit, currency, region, as_of</code> (русские/английские названия
+    столбцов узнаются) или HTML-таблица (имя в 1-й колонке, цена во 2-й). Запись проходит через
+    тот же пайплайн: нормализация → категоризация → валидация. Низкая уверенность — в очередь на проверку,
+    не в аналитику.
+  </p>
+  <form id="upload-form" style="display:grid;grid-template-columns:repeat(2,1fr);gap:10px;margin-top:8px">
+    <label><span class="muted" style="font-size:12px">Источник (id, например arbuz)</span>
+      <input name="source_id" required style="width:100%;height:34px;padding:0 10px;border-radius:8px;border:1px solid var(--kt-ai-border);background:var(--kt-ai-surface);font-size:14px"></label>
+    <label><span class="muted" style="font-size:12px">Название источника</span>
+      <input name="source_name" style="width:100%;height:34px;padding:0 10px;border-radius:8px;border:1px solid var(--kt-ai-border);background:var(--kt-ai-surface);font-size:14px"></label>
+    <label><span class="muted" style="font-size:12px">Ссылка на страницу (для прозрачности)</span>
+      <input name="source_url" style="width:100%;height:34px;padding:0 10px;border-radius:8px;border:1px solid var(--kt-ai-border);background:var(--kt-ai-surface);font-size:14px"></label>
+    <label><span class="muted" style="font-size:12px">Дата цен (YYYY-MM-DD, если в файле нет)</span>
+      <input name="as_of" style="width:100%;height:34px;padding:0 10px;border-radius:8px;border:1px solid var(--kt-ai-border);background:var(--kt-ai-surface);font-size:14px"></label>
+    <label style="grid-column:1/-1"><span class="muted" style="font-size:12px">Файл (.csv / .html)</span>
+      <input type="file" name="file" accept=".csv,.html,.htm,.txt" required style="width:100%;padding:6px;font-size:14px"></label>
+    <div style="grid-column:1/-1">
+      <button type="submit" class="kt-ai-btn" data-variant="primary">Загрузить и проанализировать</button>
+      <span id="upload-result" class="muted" style="font-size:13px;margin-left:10px"></span>
+    </div>
+  </form>
+</div>
+
+<script>
+document.getElementById('upload-form').addEventListener('submit', function(e) {
+  e.preventDefault();
+  const f = new FormData(e.target);
+  const out = document.getElementById('upload-result');
+  out.textContent = 'Загружаю…';
+  fetch('api/upload', { method: 'POST', body: f }).then(r => r.json()).then(res => {
+    if (!res.ok) { out.textContent = 'Ошибка: ' + res.error; return; }
+    out.innerHTML = 'Готово: записано <b>' + res.loaded + '</b>, в карантине <b>' +
+      res.quarantined + '</b>, пропущено ' + res.skipped +
+      (res.total ? ' / всего ' + res.total : '') +
+      ' строк. Источник: ' + res.source_id;
+  }).catch(err => { out.textContent = 'Ошибка сети: ' + err; });
+});
+</script>

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 <table> 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 <table> 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 "<table" in text[:2000].lower():
+        return parse_html(text, source_id=source_id, source_name=source_name,
+                           source_url=source_url, as_of_str=as_of)
+    # Fallback: try CSV
+    return parse_csv(text, source_id=source_id, source_name=source_name,
+                      source_url=source_url, as_of_str=as_of)

src/tests/test_mvp.py

@@ -189,6 +189,40 @@ class TestForecast:
         assert len(out["forecast"]) == 3
 
 
+class TestUpload:
+    def test_csv_roundtrip(self):
+        from src.pipeline.upload import parse_csv
+        from src import db
+        text = ("product,region,price\n"
+                "Пшеница,КЗ-АЛМА,170\n"
+                "Кукуруза,КЗ-АЛМА,150\n")
+        res = parse_csv(text, source_id="up_csv", as_of_str="2026-09-24")
+        assert res["total"] == 2
+        assert res["loaded"] + res["quarantined"] == 2
+        rows = db.exec_sql("SELECT COUNT(*) AS t FROM prices WHERE source_id='up_csv'")
+        assert rows[0]["t"] == 2
+
+    def test_html_roundtrip(self):
+        from src.pipeline.upload import parse_html
+        from src import db
+        fx = Path(__file__).parent / "fixtures" / "sharyn_sample.html"
+        text = fx.read_text(encoding="utf-8")
+        res = parse_html(text, source_id="up_html", as_of_str="2026-09-24")
+        # 4 rows of name+price; all should parse (some may quarantine on categorization)
+        assert res["loaded"] + res["quarantined"] >= 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