Thu thập dữ liệu và lập báo cáo bằng Python và AI
Thu thập CSV có kiểm soát, tính tổng bằng Python và dùng AI viết phần diễn giải chỉ dựa trên bảng đã kiểm tra.
Tổng quan
Dành cho người học muốn tự hoàn thành bài thực hành thu thập dữ liệu và lập báo cáo bằng python và ai, với dữ liệu mẫu có thể kiểm tra.
- Tách tính toán số liệu khỏi viết diễn giải bằng AI.
- Kiểm tra schema, kích thước và timeout nguồn dữ liệu.
- Tạo báo cáo Markdown có thể chạy lại.
Bạn sẽ hoàn thành: report.md với Design=3, Engineering=3, tổng 6 giờ và phần diễn giải có căn cứ.
Chuẩn bị
- Python 3.10+, trình soạn thảo; tùy chọn một URL HTTPS CSV do bạn được phép sử dụng.
Mặc định chạy offline với fixture. Tùy chọn --url đọc dữ liệu công khai/được phép, không dùng scraping hay credentials. AI chat để viết diễn giải là tùy chọn và có thể có phí/hạn mức.
Thực hành từng bước
Bước 1. Chốt schema và dữ liệu mẫu
Schema gồm team và hours, mỗi dòng biểu diễn giờ đào tạo nguyên không âm. Dùng fixture trước để biết đáp án đúng: Design có 2+1=3, Engineering có 3, tổng 6. Không dùng AI để cộng các số mà code có thể tính chính xác.
Bước 2. Viết pipeline có kiểm tra nguồn
Lưu code vào report.py. Chỉ truyền --url từ nguồn do bạn kiểm soát hoặc đã được phép đọc. Kiểm tra HTTPS và giới hạn byte giúp bài tập nhỏ gọn nhưng không đủ để biến chương trình thành dịch vụ tải URL công khai an toàn.
import argparse
import csv
import io
from pathlib import Path
from urllib.parse import urlsplit
from urllib.request import Request, urlopen
SAMPLE = "team,hours\nDesign,2\nEngineering,3\nDesign,1\n"
def read_source(url=None):
if url is None:
return SAMPLE
if urlsplit(url).scheme != "https":
raise ValueError("use an approved HTTPS CSV source")
request = Request(url, headers={"User-Agent": "TrainingReport/1.0"})
with urlopen(request, timeout=15) as response:
data = response.read(100001)
if len(data) > 100000:
raise ValueError("source exceeds 100 KB")
return data.decode("utf-8-sig")
def aggregate(text):
rows = csv.DictReader(io.StringIO(text))
if not {"team", "hours"} <= set(rows.fieldnames or []):
raise ValueError("team and hours columns required")
totals = {}
for row in rows:
team = row["team"].strip()
hours = int(row["hours"])
if not team or hours < 0:
raise ValueError("invalid team or hours")
totals[team] = totals.get(team, 0) + hours
return totals
def write_report(totals):
lines = ["# Training hours", "", "| Team | Hours |", "| --- | ---: |"]
for team, hours in sorted(totals.items()):
safe = team.replace("|", "\\|").replace("\n", " ")
lines.append(f"| {safe} | {hours} |")
lines += ["", f"Total hours: {sum(totals.values())}"]
Path("report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--url")
args = parser.parse_args()
totals = aggregate(read_source(args.url))
write_report(totals)
print("Total hours:", sum(totals.values()))Bước 3. Chạy fixture và kiểm tra số
python3 report.pyMở report.md, kiểm tra hai dòng nhóm có tổng 3/3. Thay SAMPLE bằng dữ liệu thiếu cột hoặc hours=-1: chương trình phải báo lỗi trước khi viết báo cáo mới. Giữ bản báo cáo cũ có nhãn thời điểm/nguồn bên ngoài nếu lưu lịch sử trong một hệ thật.
Bước 4. Thử một nguồn CSV được phép
Đưa fixture lên một vị trí HTTPS của bạn rồi chạy python3 report.py --url YOUR_APPROVED_HTTPS_CSV_URL sau khi thay placeholder bằng URL thật. Đối chiếu tổng với fixture. Không dùng URL ví dụ bịa, không gửi key trong query string, và không gọi lặp nhanh khi nguồn trả lỗi.
Bước 5. Thêm diễn giải AI rồi đối chiếu
Dữ liệu kiểm tra: Design=3 giờ, Engineering=3 giờ, tổng=6 giờ; đây là fixture đào tạo, không phải dữ liệu kinh doanh thật. Viết ba bullet mô tả bảng, nêu rằng chỉ có một batch nên chưa thể suy ra xu hướng, nguyên nhân hay hiệu quả. Không thay bất kỳ con số nào.Dán đoạn đã rà soát dưới bảng, lưu riêng như phần diễn giải, không ghi đè số liệu bằng văn bản model. Khi cần chạy định kỳ, dùng scheduler của hệ điều hành sau khi có log, retry có giới hạn và cách nhận biết batch đã xử lý.
Kiểm tra kết quả
- Fixture luôn cho 3/3 và tổng 6; dữ liệu sai bị từ chối.
- Diễn giải giữ nguyên số và không suy diễn xu hướng.
Xử lý lỗi thường gặp
Endpoint trả HTML thay vì CSV
Nguyên nhân: URL trỏ trang web hoặc trang đăng nhập.
Cách xử lý: Dùng đường dẫn CSV thực sự, kiểm tra header team,hours và quyền truy cập.
Báo cáo sai do dữ liệu lặp
Nguyên nhân: Schema hiện tại không có event ID để chống trùng.
Cách xử lý: Không tự đoán dòng trùng; thêm ID từ nguồn và quy tắc dedup có test trước khi dùng dữ liệu thật.
Bài tập & mở rộng
- Thêm test dữ liệu không có dòng và team chứa dấu |.
- Bổ sung event ID và log số dòng hợp lệ/bị từ chối cho một nguồn của bạn.
Không có scraping, lịch chạy tự động hay lời hứa cập nhật liên tục trong script này. Cần thiết kế chống SSRF, redirects và quyền mạng riêng trước khi nhận URL không tin cậy.
