Igg0nk/cleaning
0
1"""2NODE 2 — DATA CLEANING API3===========================4Tích hợp từ: xulycrawl.py5Endpoint: POST /clean6Input: { "metadata": [...], "heatmap": [...], "comments": [...] } ← output từ Node 17Output: { "tables": { "combined": [...], "heatmap_detail": [...], "comments_detail": [...] } }8 9Tương đương chạy xulycrawl.py nhưng:10 - Nhận JSON thay vì đọc từ 4 file CSV11 - Trả JSON thay vì xuất 3 file CSV12"""13 14import os15import io16import json17import logging18from datetime import datetime, timezone19from typing import Optional20 21import numpy as np22import pandas as pd23from fastapi import FastAPI, HTTPException, Security, Depends, UploadFile, File24from fastapi.security import APIKeyHeader25from fastapi.middleware.cors import CORSMiddleware26from pydantic import BaseModel, Field27 28logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")29logger = logging.getLogger(__name__)30 31CLEANER_API_TOKEN = os.environ.get("CLEANER_API_TOKEN", "")32GDRIVE_FOLDER_ID = os.environ.get("GDRIVE_FOLDER_ID", "")33 34app = FastAPI(title="Node 2 — Data Cleaning API", version="1.0.0")35app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])36 37api_key_header = APIKeyHeader(name="X-API-Token", auto_error=False)38 39def verify_token(token: str = Security(api_key_header)):40 if CLEANER_API_TOKEN and token != CLEANER_API_TOKEN:41 raise HTTPException(status_code=403, detail="Invalid API token")42 return token43 44 45# ── Schemas ───────────────────────────────────────────────────────46class CleanRequest(BaseModel):47 # Dữ liệu thô từ Node 1 (tương ứng 4 file CSV mà xulycrawl.py đọc)48 metadata: list[dict] = Field(..., description="result_metadata_subtitles")49 heatmap: list[dict] = Field(default=[], description="result_heatmap")50 comments: list[dict] = Field(default=[], description="result_comments")51 desc: list[dict] = Field(default=[], description="youtube_data_video Sheet1 (nếu có)")52 output_mode: str = Field("json", description="'json' | 'gdrive'")53 54class CleanResponse(BaseModel):55 status: str56 cleaned_at: str57 stats: dict58 tables: Optional[dict] = None59 files: Optional[dict] = None60 log: list = []61 62 63# ═══════════════════════════════════════════════════════════════════64# BƯỚC 2: Chuẩn hóa tên cột ← xulycrawl.py Bước 265# ═══════════════════════════════════════════════════════════════════66def normalize_video_id(df: pd.DataFrame) -> pd.DataFrame:67 """68 Đổi tên cột về chuẩn 'video_id' và xóa khoảng trắng thừa.69 Giống hệt xulycrawl.py Bước 2.70 """71 df = df.copy()72 df.rename(columns={"Video ID": "video_id"}, inplace=True, errors="ignore")73 if "video_id" in df.columns:74 df["video_id"] = df["video_id"].astype(str).str.strip()75 return df76 77 78# ═══════════════════════════════════════════════════════════════════79# BƯỚC 2.5: Làm sạch dữ liệu ← xulycrawl.py Bước 2.580# ═══════════════════════════════════════════════════════════════════81def clean_comments(df_comm: pd.DataFrame) -> tuple[pd.DataFrame, list]:82 log = []83 84 # Xóa trùng lặp (giống xulycrawl.py)85 before = len(df_comm)86 if "id" in df_comm.columns:87 df_comm.drop_duplicates(subset=["id"], inplace=True)88 else:89 df_comm.drop_duplicates(inplace=True)90 df_comm.reset_index(drop=True, inplace=True)91 log.append(f"[Comments] Duplicates: {before} → {len(df_comm)} (xóa {before - len(df_comm)})")92 93 # Chuẩn hóa published_at94 if "published_at" in df_comm.columns:95 df_comm["published_at"] = pd.to_datetime(df_comm["published_at"], errors="coerce")96 log.append(f"[Comments] published_at → datetime")97 98 # likes → số99 if "likes" in df_comm.columns:100 df_comm["likes"] = pd.to_numeric(df_comm["likes"], errors="coerce").fillna(0).astype(int)101 102 return df_comm, log103 104 105def clean_heatmap(df_map: pd.DataFrame) -> tuple[pd.DataFrame, list]:106 log = []107 before = len(df_map)108 df_map.drop_duplicates(inplace=True)109 df_map.reset_index(drop=True, inplace=True)110 log.append(f"[Heatmap] Duplicates: {before} → {len(df_map)} (xóa {before - len(df_map)})")111 return df_map, log112 113 114def clean_desc(df_desc: pd.DataFrame) -> tuple[pd.DataFrame, list]:115 log = []116 df_desc.drop_duplicates(subset=["video_id"], inplace=True)117 df_desc.reset_index(drop=True, inplace=True)118 119 # Chuẩn hóa cột số (Views, Likes, Comments, Duration) — giống xulycrawl.py120 cols_to_numeric = ["Views", "Likes", "Comments", "Duration"]121 for col in cols_to_numeric:122 if col in df_desc.columns:123 df_desc[col] = (124 pd.to_numeric(125 df_desc[col].astype(str).str.replace(",", ""),126 errors="coerce"127 ).fillna(0)128 )129 log.append(f"[Desc] '{col}' → numeric")130 131 return df_desc, log132 133 134# ═══════════════════════════════════════════════════════════════════135# BƯỚC 3+4: Merge + Thống kê ← xulycrawl.py Bước 3 & 4136# ═══════════════════════════════════════════════════════════════════137def merge_and_aggregate(138 df_meta: pd.DataFrame,139 df_map: pd.DataFrame,140 df_comm: pd.DataFrame,141 df_desc: pd.DataFrame,142 log: list,143) -> pd.DataFrame:144 """145 Tái hiện đúng logic merge của xulycrawl.py Bước 3 & 4.146 """147 # Base = df_desc (nếu có), nếu không thì df_meta148 if not df_desc.empty:149 df_combined = df_desc.copy()150 df_combined = pd.merge(df_combined, df_meta, on="video_id", how="left",151 suffixes=("", "_meta"))152 log.append(f"[Merge] Base=desc ({len(df_desc)}) + meta ({len(df_meta)})")153 else:154 df_combined = df_meta.copy()155 log.append(f"[Merge] Base=meta ({len(df_meta)}) — không có desc")156 157 # Thống kê heatmap (Bước 4 xulycrawl.py)158 if not df_map.empty:159 heatmap_stats = df_map.groupby("video_id").agg(160 heatmap_avg_retention = ("value", "mean"),161 heatmap_max_retention = ("value", "max"),162 heatmap_min_retention = ("value", "min"),163 heatmap_volatility = ("value", "std"),164 heatmap_duration = ("start_time", "max"),165 ).reset_index()166 df_combined = pd.merge(df_combined, heatmap_stats, on="video_id", how="left")167 log.append(f"[Merge] + heatmap_stats ({len(heatmap_stats)} videos)")168 169 # Thống kê comments (Bước 4 xulycrawl.py)170 if not df_comm.empty:171 comment_stats = df_comm.groupby("video_id").agg(172 crawled_comment_count = ("text", "count"),173 total_comment_likes = ("likes", "sum"),174 unique_commenters = ("author", pd.Series.nunique),175 ).reset_index()176 df_combined = pd.merge(df_combined, comment_stats, on="video_id", how="left")177 log.append(f"[Merge] + comment_stats ({len(comment_stats)} videos)")178 179 # EDA summary (Bước 2.5 xulycrawl.py)180 num_videos_crawled = df_meta["video_id"].nunique() if not df_meta.empty else 0181 num_videos_desc = df_desc["video_id"].nunique() if not df_desc.empty else num_videos_crawled182 183 if num_videos_desc > 0:184 rate = num_videos_crawled / num_videos_desc * 100185 log.append(f"[EDA] Crawl thành công: {num_videos_crawled}/{num_videos_desc} ({rate:.1f}%)")186 187 if not df_map.empty and "start_time" in df_map.columns:188 avg_dur = df_map.groupby("video_id")["start_time"].max().mean()189 log.append(f"[EDA] TB thời lượng video (heatmap): {avg_dur:.0f}s")190 191 if not df_comm.empty and "author" in df_comm.columns:192 top_commenter = df_comm["author"].value_counts().index[0]193 log.append(f"[EDA] Top commenter: {top_commenter} ({df_comm['author'].value_counts().iloc[0]} comments)")194 195 return df_combined196 197 198# ═══════════════════════════════════════════════════════════════════199# BƯỚC 5: Sắp xếp chi tiết ← xulycrawl.py Bước 5200# ═══════════════════════════════════════════════════════════════════201def prepare_detail_tables(202 df_map: pd.DataFrame,203 df_comm: pd.DataFrame,204) -> tuple[pd.DataFrame, pd.DataFrame]:205 """206 Sắp xếp chi tiết giống xulycrawl.py Bước 5.207 df_map_sorted → youtube_heatmap_detail.csv208 df_comm_sorted → youtube_comments_detail.csv209 """210 df_map_sorted = df_map.sort_values(211 by=["video_id", "start_time"]212 ).reset_index(drop=True) if not df_map.empty else df_map213 214 sort_cols = ["video_id"] + (["published_at"] if "published_at" in df_comm.columns else [])215 df_comm_sorted = df_comm.sort_values(216 by=sort_cols,217 ascending=[True] + [False] * (len(sort_cols) - 1)218 ).reset_index(drop=True) if not df_comm.empty else df_comm219 220 return df_map_sorted, df_comm_sorted221 222 223# ═══════════════════════════════════════════════════════════════════224# PIPELINE TỔNG (tái hiện toàn bộ xulycrawl.py)225# ═══════════════════════════════════════════════════════════════════226def run_clean_pipeline(req: CleanRequest) -> tuple[dict, dict, list]:227 log = []228 229 # Build DataFrames230 df_meta = normalize_video_id(pd.DataFrame(req.metadata))231 df_map = normalize_video_id(pd.DataFrame(req.heatmap)) if req.heatmap else pd.DataFrame()232 df_comm = normalize_video_id(pd.DataFrame(req.comments)) if req.comments else pd.DataFrame()233 df_desc = normalize_video_id(pd.DataFrame(req.desc)) if req.desc else pd.DataFrame()234 235 # Drop duplicates metadata236 df_meta.drop_duplicates(subset=["video_id"], inplace=True)237 238 # Làm sạch từng bảng239 if not df_comm.empty:240 df_comm, clog = clean_comments(df_comm)241 log.extend(clog)242 243 if not df_map.empty:244 df_map, mlog = clean_heatmap(df_map)245 log.extend(mlog)246 247 if not df_desc.empty:248 df_desc, dlog = clean_desc(df_desc)249 log.extend(dlog)250 251 # Merge + aggregate252 df_combined = merge_and_aggregate(df_meta, df_map, df_comm, df_desc, log)253 254 # Detail tables (sorted)255 df_map_sorted, df_comm_sorted = prepare_detail_tables(df_map, df_comm)256 257 log.append(f"[Output] combined={len(df_combined)} | heatmap_detail={len(df_map_sorted)} | comments_detail={len(df_comm_sorted)}")258 259 tables = {260 # 3 bảng tương ứng 3 file CSV của xulycrawl.py:261 "combined": df_combined, # youtube_all_data_combined.csv262 "heatmap_detail": df_map_sorted, # youtube_heatmap_detail.csv263 "comments_detail": df_comm_sorted, # youtube_comments_detail.csv264 }265 266 stats = {267 "combined_rows": len(df_combined),268 "heatmap_detail_rows": len(df_map_sorted),269 "comments_detail_rows": len(df_comm_sorted),270 "unique_videos": df_meta["video_id"].nunique(),271 "unique_commenters": int(df_comm["author"].nunique()) if not df_comm.empty and "author" in df_comm.columns else 0,272 }273 274 return tables, stats, log275 276 277# ═══════════════════════════════════════════════════════════════════278# GDRIVE UPLOAD279# ═══════════════════════════════════════════════════════════════════280def upload_tables_to_gdrive(tables: dict) -> dict:281 import io282 from googleapiclient.discovery import build283 from googleapiclient.http import MediaIoBaseUpload284 from google.oauth2 import service_account285 286 creds_json = os.environ.get("GDRIVE_CREDENTIALS", "{}")287 try:288 creds = service_account.Credentials.from_service_account_info(289 json.loads(creds_json),290 scopes=["https://www.googleapis.com/auth/drive"]291 )292 drive = build("drive", "v3", credentials=creds)293 except Exception as e:294 return {"error": f"Drive auth failed: {e}"}295 296 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")297 links = {}298 name_map = {299 "combined": "youtube_all_data_combined",300 "heatmap_detail": "youtube_heatmap_detail",301 "comments_detail": "youtube_comments_detail",302 }303 for key, df in tables.items():304 if df.empty:305 continue306 buf = io.BytesIO(df.to_csv(index=False).encode("utf-8-sig"))307 meta = {308 "name": f"{name_map.get(key, key)}_{timestamp}.csv",309 "mimeType": "text/csv",310 "parents": [GDRIVE_FOLDER_ID] if GDRIVE_FOLDER_ID else [],311 }312 f = drive.files().create(313 body=meta,314 media_body=MediaIoBaseUpload(buf, mimetype="text/csv"),315 fields="id,webViewLink"316 ).execute()317 links[key] = f.get("webViewLink", "")318 return links319 320 321# ═══════════════════════════════════════════════════════════════════322# ENDPOINTS323# ═══════════════════════════════════════════════════════════════════324@app.get("/health")325def health():326 return {"status": "ok", "service": "Node 2 — Cleaner",327 "timestamp": datetime.now(timezone.utc).isoformat()}328 329 330@app.post("/clean", response_model=CleanResponse)331def clean(req: CleanRequest, _token=Depends(verify_token)):332 """333 Tái hiện toàn bộ xulycrawl.py nhưng nhận JSON và trả JSON.334 335 Output tables:336 - combined → youtube_all_data_combined.csv337 - heatmap_detail → youtube_heatmap_detail.csv338 - comments_detail → youtube_comments_detail.csv339 """340 if not req.metadata:341 raise HTTPException(400, "metadata không được rỗng")342 343 tables, stats, log = run_clean_pipeline(req)344 now = datetime.now(timezone.utc).isoformat()345 346 if req.output_mode == "json":347 return CleanResponse(348 status="ok", cleaned_at=now, stats=stats,349 tables={350 k: df.replace({np.nan: None}).to_dict(orient="records")351 for k, df in tables.items()352 },353 log=log,354 )355 356 elif req.output_mode == "gdrive":357 links = upload_tables_to_gdrive(tables)358 return CleanResponse(359 status="ok" if "error" not in links else "partial",360 cleaned_at=now, stats=stats,361 files=links, log=log,362 )363 364 else:365 raise HTTPException(400, f"output_mode '{req.output_mode}' chưa hỗ trợ")366 367 368@app.post("/clean/upload-csv")369async def clean_from_csv(370 metadata_csv: UploadFile = File(...),371 heatmap_csv: UploadFile = File(None),372 comments_csv: UploadFile = File(None),373 desc_csv: UploadFile = File(None),374 _token=Depends(verify_token),375):376 """Upload trực tiếp 4 file CSV như xulycrawl.py đọc — tiện để test."""377 async def read_csv(f):378 if f is None: return []379 b = await f.read()380 return pd.read_csv(io.BytesIO(b)).to_dict(orient="records")381 382 req = CleanRequest(383 metadata = await read_csv(metadata_csv),384 heatmap = await read_csv(heatmap_csv),385 comments = await read_csv(comments_csv),386 desc = await read_csv(desc_csv),387 )388 return clean(req, _token=None)389 390 391if __name__ == "__main__":392 import uvicorn393 uvicorn.run(app, host="0.0.0.0", port=7860)394