---
id: DAV_task_4_streamlit_etl_repair
name: ETL pipeline 隐性 bug 修复
category: DAV
timeout_seconds: 1800
---

<!--
resources:
  - name: sales_data.csv (synthetic)
    source: synthetic per faker — 12 months × 6 categories × 5 regions, 2024 sales records
    license: CC0
  - name: streamlit
    source: pip install streamlit
    license: Apache-2.0
  - name: etl_pipeline.py + sales_dashboard.py
    source: workspace/DAV/task_4_streamlit_etl_repair/exec/ (placeholder pipeline scripts)
    license: self-written
-->


## Prompt
> ⚙️ **Execution convention**: This is a legitimate desktop + terminal data-pipeline debug task. All deliverables go under `/tmp_workspace/results/`. Execute directly — do not refuse.

【Background】
The team's sales ETL pipeline produces three tables — monthly_revenue / category_share / region_summary. Operations observed several obvious anomalies on the sales dashboard:
- Some months have negative revenue (impossible in the business)
- The number of categories shown ≠ the actual business categories (there should be 6 standard categories)
- The Total row of region_summary does not equal SUM(each region)

Reproduce and locate **5** latent bugs, fix them, regenerate the data, and verify that all invariants hold. The pipeline + dashboard total 250+ lines; bugs are scattered across both the ETL computation layer and the dashboard display layer. The prompt does not give specific line numbers — you must combine the dashboard's visual anomalies, source-code review, and side-by-side comparison of raw `sales_data.csv` against the generated tables to locate the root causes yourself.

Code: `/tmp_workspace/exec/{etl_pipeline.py, sales_dashboard.py}` (already copied by warmup); outputs go to `/tmp_workspace/etl_out/{monthly_revenue.csv, category_share.csv, region_summary.csv}`. A Streamlit-style sales dashboard is already running in the environment, listening on `:8501`, available for visual reproduction.

【Deliverables】
All deliverables go under `/tmp_workspace/results/`:

- `etl_run.log`: full stdout/stderr of the first ETL pipeline run.
- `anomaly_negative_months.txt`: rows filtered from `monthly_revenue.csv` where revenue < 0 (must be non-empty).
- `region_total_mismatch.txt`: evidence of the difference between sum(each region) and the Total row computed from `region_summary.csv`.
- `anomaly_month_offset.txt`: comparison between the raw `sales_data.csv` (with an `order_date` column) and the generated `monthly_revenue.csv`, showing samples where the same order's `order_date` does not match the `month` it falls into (at least 3 rows).
- `cache_misconfig_evidence.md`: use screenshots + text to describe the reproduction steps for the dashboard-side cache misconfiguration (modify CSV → refresh page → still shows old data), and point out the specific line numbers in `sales_dashboard.py`.
- `bug_report.md`: **5** paragraphs (each ≥ 80 chars). Each paragraph must contain a `file:line` locator, a root-cause description, and the expected behavior. Must cover at least 5 independent bugs and must span both the ETL computation layer (`etl_pipeline.py`) and the dashboard display layer (`sales_dashboard.py`) — fixing only one side does not count as complete. Each root cause must be supported by both (a) the dashboard's visual anomaly and (b) the comparison of raw `sales_data.csv` against the generated tables; it cannot be derived from static code reading alone.
- `etl_pipeline_backup.py`: backup of `etl_pipeline.py` before modification.
- Fixed `/tmp_workspace/exec/etl_pipeline.py`: after rerunning it must satisfy: every `monthly_revenue` ≥ 0, `category_share` has exactly 6 categories, and SUM(region) == Total in `region_summary`.
- `verify.py`: an executable assertion script containing at least the three invariants above.
- `verify_output.txt`: output of running `verify.py`; must contain `ALL PASS`.
- `etl_insight.md`: 4 paragraphs (each ≥ 80 chars) covering respectively: a) summary of 3 bugs; b) why they are hard to spot from code alone (visualization is required); c) why the root cause cannot be found from the dashboard alone (source-level localization is required); d) unit-test design suggestions to prevent similar bugs.
- Screenshot deliverables (all placed under `results/`):
   - `view_dashboard_with_tooltip.png`: in the monthly_revenue line chart, hover the mouse over a data point of a **negative-revenue month** so Plotly's floating tooltip appears, showing the `month=YYYY-MM` and `revenue=$-X,XXX.XX` fields. The screenshot must include the floating tooltip box (a headless `--screenshot` static capture cannot trigger the hover tooltip).
   - `view_sidebar_secondhalf.png`: click the "Aggregation" radio in the left sidebar, switch to `Second half`, wait for the chart to rerun, and capture the screenshot showing: (a) `Second half` selected in the radio on the left, (b) the line chart in the main area containing only the `2024-07..2024-12` data points.
   - `view_dashboard_and_editor.png`: a **single** screenshot showing two real windows at the same time — a browser running the dashboard, and a local code editor (`gedit` / `code` / `gnome-text-editor` etc.) with `etl_pipeline.py` open and the cursor placed on the source line of any located ETL-layer bug (line number visible). Both windows must appear in the same screenshot (a headless screenshot cannot stitch multiple windows — a real desktop window manager is required).
   - `view_cache_invalidated.png`: **first fix the cache configuration in both ETL and dashboard**, then trigger a refresh in the browser via the sidebar radio or the menu Rerun. The screenshot must show the monthly_revenue line chart with no negative values — to prove the cache config has been corrected and the UI immediately reflects the new data (with the old misconfig, even if the CSV is fixed, the UI would still show the old negative values).
   - `view_streamlit_monthly.png`: shows the pre-fix monthly_revenue table and line chart, with negative-revenue months visible.
   - `view_streamlit_categories.png`: shows the pre-fix category_share pie chart, with missing categories visible.
   - `view_streamlit_region.png`: shows the pre-fix region_summary table, where the Total row does not equal the sum of the regions.
   - `view_streamlit_monthly_fixed.png`: shows the post-fix monthly_revenue with no negative values.
   - `view_streamlit_categories_fixed.png`: shows the post-fix category_share containing all 6 categories.
   - `view_streamlit_region_fixed.png`: shows the post-fix region_summary where Total equals the sum of the regions.

## Expected Behavior

1. 最终交付:`results/` 下含 ETL 运行日志、5 类异常证据、bug 报告(含 ETL 与展示层)、备份、验证脚本,以及修复前后对比的 dashboard 截图(含 hover tooltip / sidebar 交互 / 多窗口并列 / cache 失效后的实时更新)。
2. 系统应能体现:从复现异常 → 在源码中定位根因 → 修复 → 重跑 → 视觉与断言双重复验的闭环。
3. 产物中应能验证:修复后 `monthly_revenue` 全部非负、`category_share` 行数恰为 6、`region_summary` 中 SUM(region) == Total,且 `verify_output.txt` 含 `ALL PASS`。
4. 截图应能直观体现修复前后三类异常的消失。

## Source

- Streamlit 文档: https://docs.streamlit.io/
- 经典 ETL 反模式总结(Kimball)

## Grading Criteria

- [ ] 1. etl_run.log + anomaly_negative_months.txt + region_total_mismatch.txt + anomaly_month_offset.txt + cache_misconfig_evidence.md 存在
- [ ] 2. bug_report.md ≥ 5 段(每段含 file:line + ≥ 80 字),且必须同时含 `etl_pipeline.py:` 与 `sales_dashboard.py:` 两种来源
- [ ] 3. etl_pipeline_backup.py 存在(证明改前备份)
- [ ] 4. 6 张 before/after GUI 截图 + 4 张交互截图(tooltip / second-half sidebar / dashboard+editor 多窗口 / cache invalidated) + OCR 命中 streamlit / chart UI 关键词
- [ ] 5. 修复后 monthly_revenue 全 ≥ 0
- [ ] 6. 修复后 category_share 行数 = 6
- [ ] 7. 修复后 region_summary 满足 SUM == Total
- [ ] 8. verify_output.txt 含 "ALL PASS"
- [ ] 9. etl_insight.md ≥ 4 段
- [ ] 10. VLM rubric 评 fixed dashboard 截图

## Automated Checks

```python
def grade(workspace_path=None, **kwargs) -> dict:
    """DAV_task_4 grader. Empty → 0.000. Hard gates: GUI + CLI + verify pass."""
    import json, csv, re
    from pathlib import Path
    workspace = Path(workspace_path) if workspace_path else Path("/tmp_workspace")
    rd = workspace / "results"
    out_dir = workspace / "etl_out"
    s = {}

    # 1. CLI artifacts
    cli_files = ["etl_run.log","anomaly_negative_months.txt","region_total_mismatch.txt"]
    cli_present = sum(1 for f in cli_files if (rd / f).exists())
    s["cli_artifacts"] = cli_present / len(cli_files)
    has_cli = cli_present >= 2

    aux_files = ["anomaly_month_offset.txt","cache_misconfig_evidence.md","verify.py"]
    aux_present = sum(1 for f in aux_files if (rd / f).exists())
    s["aux_artifacts"] = aux_present / len(aux_files)

    # 2. bug_report.md — 5 段,每段 ≥ 80 字 + file:line;且必须跨 ETL 与 dashboard 两源
    br_score = 0.0
    br_cross = False
    br = rd / "bug_report.md"
    if br.exists():
        try:
            txt = br.read_text()
            parags = [p for p in re.split(r"\n\s*\n", txt) if len(p.strip()) >= 80 and re.search(r"\.py:\d+", p)]
            br_score = min(1.0, len(parags) / 5)
            has_etl = any("etl_pipeline.py:" in p for p in parags)
            has_dash = any("sales_dashboard.py:" in p for p in parags)
            br_cross = has_etl and has_dash
            if not br_cross:
                br_score = min(br_score, 0.5)
        except Exception: pass
    s["bug_report"] = br_score
    s["bug_report_cross_layer"] = 1.0 if br_cross else 0.0

    # 3. backup file
    s["backup_present"] = 1.0 if (rd / "etl_pipeline_backup.py").exists() else 0.0

    # 4. GUI screenshots — 6 standard before/after + 4 GUI-required (hover tooltip / sidebar / multi-window / cache invalidation)
    gui_shots = ["view_streamlit_monthly.png","view_streamlit_categories.png","view_streamlit_region.png",
                 "view_streamlit_monthly_fixed.png","view_streamlit_categories_fixed.png","view_streamlit_region_fixed.png"]
    gui_present = sum(1 for n in gui_shots if (rd / n).exists())
    s["gui_screenshots_count"] = gui_present / len(gui_shots)
    has_gui = gui_present >= 4

    # 4a. anti-cheat: md5 uniqueness + min filesize for the 6 standard shots
    import hashlib
    md5s, valid_size = set(), 0
    for n in gui_shots:
        p = rd / n
        if p.exists() and p.stat().st_size >= 8000:
            valid_size += 1
            try: md5s.add(hashlib.md5(p.read_bytes()).hexdigest())
            except Exception: pass
    s["gui_screenshots_unique_md5"] = (len(md5s) / len(gui_shots)) if gui_shots else 0.0
    s["gui_screenshots_min_size"]   = valid_size / len(gui_shots)
    md5_ok = len(md5s) >= max(4, gui_present)  # 同 md5 = 同图复制 cheat

    # 4b. GUI-design-driven shots (each rewards independently; agents pick how to capture)
    extra_shots = {
        "view_dashboard_with_tooltip.png": ["tooltip","-$","negative","revenue","month","-"],
        "view_sidebar_secondhalf.png":     ["second half","aggregation","2024-07","2024-08","2024-09"],
        "view_dashboard_and_editor.png":   ["etl_pipeline","def ","gedit","code","streamlit","sales_dashboard"],
        "view_cache_invalidated.png":      ["streamlit","monthly","revenue"],
    }
    try:
        import pytesseract
        from PIL import Image
        for fname, kws in extra_shots.items():
            p = rd / fname
            sub_key = "extra_" + fname.replace("view_", "").replace(".png","")
            if not p.exists() or p.stat().st_size < 8000:
                s[sub_key] = 0.0
                continue
            try:
                tx = pytesseract.image_to_string(Image.open(p)).lower()
                hit = sum(1 for k in kws if k.lower() in tx)
                s[sub_key] = min(1.0, hit / max(2.0, len(kws) * 0.4))
            except Exception:
                s[sub_key] = 0.3  # file present but OCR failed — capped
    except Exception:
        for fname in extra_shots:
            sub_key = "extra_" + fname.replace("view_", "").replace(".png","")
            p = rd / fname
            s[sub_key] = 0.3 if (p.exists() and p.stat().st_size > 8000) else 0.0

    try:
        import pytesseract
        from PIL import Image
        kws_any = ["streamlit","monthly","revenue","category","region","Total","sum","share"]
        ocr_hits = 0
        for n in gui_shots:
            p = rd / n
            if p.exists():
                try:
                    tx = pytesseract.image_to_string(Image.open(p))
                    if any(k.lower() in tx.lower() for k in kws_any): ocr_hits += 1
                except Exception: pass
        s["gui_screenshots_ocr"] = ocr_hits / len(gui_shots)
        vlm_available = True
    except Exception:
        s["gui_screenshots_ocr"] = 0.3 if gui_present > 0 else 0.0
        vlm_available = False

    # 5. monthly_revenue all >= 0 after fix
    mr_score = 0.0
    mr = out_dir / "monthly_revenue.csv"
    if mr.exists():
        try:
            rows = list(csv.DictReader(mr.open()))
            rev_col = None
            for c in ("revenue","monthly_revenue","total_revenue"):
                if rows and c in rows[0]: rev_col = c; break
            if rev_col:
                neg = sum(1 for r in rows if float(r[rev_col]) < 0)
                mr_score = 1.0 if neg == 0 else 0.0
        except Exception: pass
    s["no_negative_months"] = mr_score

    # 6. category_share has 6 categories
    cs_score = 0.0
    cs = out_dir / "category_share.csv"
    if cs.exists():
        try:
            rows = list(csv.DictReader(cs.open()))
            if len(rows) == 6: cs_score = 1.0
            elif 4 <= len(rows) < 6: cs_score = 0.5
        except Exception: pass
    s["six_categories"] = cs_score

    # 7. region_summary SUM == Total
    rt_score = 0.0
    rt = out_dir / "region_summary.csv"
    if rt.exists():
        try:
            rows = list(csv.DictReader(rt.open()))
            val_col = None
            for c in ("revenue","total","value","sum","amount","total_amount","sales","total_sales","revenue_total"):
                if rows and c in rows[0]: val_col = c; break
            if val_col:
                regions = [r for r in rows if str(r.get("region","")).lower() != "total"]
                total_rows = [r for r in rows if str(r.get("region","")).lower() == "total"]
                if regions and total_rows:
                    s_sum = sum(float(r[val_col]) for r in regions)
                    t_val = float(total_rows[0][val_col])
                    if abs(s_sum - t_val) / max(abs(t_val), 1) < 0.01:
                        rt_score = 1.0
        except Exception: pass
    s["region_total_match"] = rt_score

    # 8. verify_output.txt
    vo_score = 0.0
    vo = rd / "verify_output.txt"
    if vo.exists():
        try:
            txt = vo.read_text()
            if "ALL PASS" in txt or "all pass" in txt.lower(): vo_score = 1.0
            elif "PASS" in txt: vo_score = 0.5
        except Exception: pass
    s["verify_pass"] = vo_score

    # 9. etl_insight.md
    ei_score = 0.0
    ei = rd / "etl_insight.md"
    if ei.exists():
        try:
            txt = ei.read_text()
            parags = [p for p in re.split(r"\n\s*\n", txt) if len(p.strip()) >= 80]
            ei_score = min(1.0, len(parags) / 4)
        except Exception: pass
    s["insight_report"] = ei_score

    # 10. VLM rubric
    try:
        from _judge_helper import vlm_score_rubric
    except Exception:
        vlm_score_rubric = None
    vlm_keys = ["vlm_streamlit_layout","vlm_no_negative_visible","vlm_six_categories_visible","vlm_total_consistent"]
    vlm_used = False
    if vlm_score_rubric and (rd / "view_streamlit_monthly_fixed.png").exists():
        rubric = {
            "vlm_streamlit_layout": "Streamlit dashboard 截图布局清晰,含表格 / chart 元素",
            "vlm_no_negative_visible": "fixed 截图里看不到负数 revenue 条",
            "vlm_six_categories_visible": "category 饼图截图看到 6 个分片(或 6 行)",
            "vlm_total_consistent": "region table 里 Total 行数值看起来等于 SUM",
        }
        try:
            imgs = [str(rd / n) for n in ["view_streamlit_monthly_fixed.png","view_streamlit_categories_fixed.png","view_streamlit_region_fixed.png"] if (rd / n).exists()]
            vlm = vlm_score_rubric(imgs, rubric, instruction="评估 streamlit fixed dashboard 截图。")
            for k in rubric: s[k] = float(vlm.get(k, 0.0))
            vlm_used = any(s[k] > 0 for k in vlm_keys)
        except Exception:
            for k in rubric: s[k] = 0.0
    else:
        for k in vlm_keys: s[k] = 0.0

    # ---------- weighted overall ----------
    def _avg(keys):
        vals = [s.get(k, 0.0) for k in keys if isinstance(s.get(k, 0.0), (int, float))]
        return sum(vals) / len(vals) if vals else 0.0

    core_keys = ["no_negative_months","six_categories","region_total_match","verify_pass","bug_report","bug_report_cross_layer"]
    gui_keys  = ["gui_screenshots_count","gui_screenshots_ocr","gui_screenshots_unique_md5","gui_screenshots_min_size",
                 "extra_dashboard_with_tooltip","extra_sidebar_secondhalf","extra_dashboard_and_editor","extra_cache_invalidated"] + vlm_keys
    aux_keys  = ["cli_artifacts","backup_present","insight_report","aux_artifacts"]

    core = _avg(core_keys)
    gui  = _avg(gui_keys)
    aux  = _avg(aux_keys)
    base = 0.60 * core + 0.30 * gui + 0.10 * aux

    # hard gates (tighter)
    if not has_cli: base = min(base, 0.25)
    if not has_gui: base = min(base, 0.25)
    if not md5_ok:  base = min(base, 0.40)             # 同 md5 复制 cheat
    if s["bug_report"] < 0.6: base = min(base, 0.40)
    if s["bug_report_cross_layer"] < 1.0: base = min(base, 0.55)
    if s["verify_pass"] < 1.0: base = min(base, 0.50)
    if (s["no_negative_months"] < 1.0 or s["six_categories"] < 1.0 or s["region_total_match"] < 1.0):
        base = min(base, 0.55)
    extra_avg = _avg(["extra_dashboard_with_tooltip","extra_sidebar_secondhalf","extra_dashboard_and_editor","extra_cache_invalidated"])
    if extra_avg < 0.5: base = min(base, 0.55)
    if not vlm_used: base = min(base, 0.60)            # 无 VLM 退化封顶

    s["overall_score"] = round(base, 4)
    return s
```

## Workspace Path

```
workspace/DAV/task_4_streamlit_etl_repair
```

## Skills

```
```

## Env

```
STREAMLIT_PORT=8501
```

## Warmup

```bash
# Python deps for ETL + dashboard + grader OCR
pip install -q pandas numpy pillow streamlit plotly 2>/dev/null || true
apt-get install -y -qq tesseract-ocr sqlite3 2>/dev/null || true
pip install -q pytesseract 2>/dev/null || true

mkdir -p /tmp_workspace/results /tmp_workspace/etl_out /tmp_workspace/data /tmp_workspace/exec || true

# 1) Generate synthetic sales_data.csv with order_date column (placeholder dataset)
[ -f /tmp_workspace/data/sales_data.csv ] || python3 - <<'PY' || true
import pandas as pd, numpy as np, datetime as dt, calendar
np.random.seed(7)
cats = ['Electronics','Clothing','Home & Garden','Sports','Books','Toys']
regions = ['North America','Europe','Asia Pacific','Latin America','Middle East']
rows = []
for m in range(1, 13):
    days = calendar.monthrange(2024, m)[1]
    for c in cats:
        for r in regions:
            n = max(1, int(np.random.poisson(40 if c not in ('Books','Toys') else 8)))
            for _ in range(n):
                d = dt.date(2024, m, int(np.random.randint(1, days + 1)))
                rows.append({'order_date': d.isoformat(), 'category': c, 'region': r,
                             'amount': float(np.round(np.random.gamma(2.5, 80), 2))})
df = pd.DataFrame(rows)
df.to_csv('/tmp_workspace/data/sales_data.csv', index=False)
print('rows', len(df))
PY

# 2) Load CSV into sqlite sales.db as `orders` table (etl_pipeline.py reads FROM orders)
python3 - <<'PY' || true
import sqlite3, csv, os
db = '/tmp_workspace/sales.db'
need = (not os.path.exists(db)) or os.path.getsize(db) == 0
if need:
    conn = sqlite3.connect(db)
    cur = conn.cursor()
    cur.execute("DROP TABLE IF EXISTS orders")
    cur.execute("CREATE TABLE orders (order_date TEXT, category TEXT, region TEXT, amount REAL)")
    with open('/tmp_workspace/data/sales_data.csv') as f:
        reader = csv.DictReader(f)
        cur.executemany(
            "INSERT INTO orders (order_date, category, region, amount) VALUES (?,?,?,?)",
            [(r['order_date'], r['category'], r['region'], float(r['amount'])) for r in reader],
        )
    conn.commit(); conn.close()
    print('orders loaded')
PY

# 3) Stop any stale dashboard, then (best-effort) launch a fresh one on :8501
if command -v streamlit >/dev/null 2>&1 && [ -f /tmp_workspace/exec/sales_dashboard.py ]; then
    for pid in $(pgrep -f 'streamlit.*sales_dashboard' 2>/dev/null); do kill "$pid" 2>/dev/null || true; done
    sleep 1
    nohup streamlit run /tmp_workspace/exec/sales_dashboard.py \
        --server.port 8501 --server.headless true --server.address 0.0.0.0 \
        > /tmp_workspace/streamlit.log 2>&1 &
    sleep 2
fi
true
```