--- name: data-cleaner-ai label: Data Cleaner AI version: 1.0.0 language: Python runtime: subprocess (scripts/main.py) trigger_words: - data cleaning - deduplication - spreadsheet cleanup - data merge - format standardization - CRM data cleanup - Excel cleaning - clean data - remove duplicates - merge data --- # Data Cleaner AI Upload messy data — get clean, structured output. Supports multi-format parsing, AI field identification, intelligent dedup/fill/formatting, multi-source join, and Feishu-native output (Bitable + quality report doc). **Use cases:** E-commerce order cleanup, CRM customer data cleansing, bank statement reconciliation, roster cleanup, multi-system data merge. --- ## Capabilities ### F1 · Multi-Format Parsing - Excel (.xlsx / .ls) - CSV / TSV - JSON (semi-structured) - Clipboard paste text ### F2 · Smart Field Identification - AI auto-detects: name, phone, email, address, amount, date, SKU, order ID, ID number, gender, etc. - Supports user-defined field mapping override ### F3 · Data Cleaning - **Deduplication**: Exact match + fuzzy dedup (FuzzyWuzzy, threshold 88%) - **Missing value fill**: Mean / mode / semantic inference / leave blank - **Format standardization**: - Phone → `1xx-xxxx-xxxx` - Date → `YYYY-MM-DD` - Amount → 2 decimal places - Address → Province/City/District/Street standardization ### F4 · Data Classification / Tagging (PRO) - 8 built-in business rules (high-value customer, dormant user, VIP, enterprise, etc.) - Supports custom JSON rules - AI auto-tagging (requires PRO + AI API Key) ### F5 · Multi-Source Join / Merge (PRO) - Cross-file relational join on key fields - Fuzzy join when exact key not available (FuzzyWuzzy) - Conflicted field resolution: priority by source order or latest timestamp ### F6 · Feishu Native Output - Excel / CSV export - Feishu Bitable (multi-dimensional table) write-back - Data quality report auto-generated as Feishu Doc (Markdown) --- ## Tier Feature Matrix | Feature | FREE | PRO | |---------|:----:|:---:| | Multi-format parsing | ✅ | ✅ | | Basic dedup | ✅ | ✅ | | Smart fill | ❌ | ✅ | | Format standardization | ❌ | ✅ | | Fuzzy dedup | ❌ | ✅ | | Multi-source merge | ❌ | ✅ | | AI classification | ❌ | ✅ | | Data quality report | ❌ | ✅ | | Feishu Bitable output | ❌ | ✅ | --- ## Pricing **Per-call billing (no monthly fee):** | Tier | Price per Call | |------|---------------| | FREE | $0.00 USDT | | PRO | $0.01 USDT | Each cleaning pipeline execution (clean or merge) = one billable call. --- ## Usage ### Feishu Trigger ``` data cleaning deduplication spreadsheet cleanup CRM data cleanup Excel cleaning ``` ### CLI ```bash python scripts/main.py clean -i data.xlsx -o cleaned.xlsx python scripts/main.py clean -t "name,phone\nJohn,13800138000" -f csv -o cleaned.csv python scripts/main.py merge --sources customers.xlsx orders.csv --on phone -o merged.xlsx ``` ### Python API ```python from main import run_clean_pipeline result = run_clean_pipeline( sources=["orders.xlsx"], output_format="xlsx", output_path="/tmp/cleaned.xlsx", dedup_strategy="auto", fill_strategy="auto", classify=True, ai_model="deepseek", generate_report=True, ) ``` --- ## Directory Structure ``` data-cleaner-ai/ ├── SKILL.md └── scripts/ ├── main.py # Entry: run_clean_pipeline / run_merge_pipeline ├── parser.py # F1: Multi-format parsing ├── field_identifier.py # F2: AI field identification ├── cleaner.py # F3: Cleaning engine ├── classifier.py # F4: Classification / tagging ├── merger.py # F5: Multi-source join ├── reporter.py # F6: Quality report generation ├── output.py # F6: Output (Excel/CSV/Bitable/Feishu Doc) ├── tier_limits.py # Tier access control └── billing.py # SkillPay billing integration ``` --- ## Billing This skill uses **SkillPay** (skillpay.me) for per-call billing. **Fee:** $0.0100 USDT per execution (all paid tiers) **External API:** `https://skillpay.me/api/v1/billing` **Data transmitted:** User identifier (`FEISHU_USER_ID` environment variable) Billing occurs at the start of each cleaning or merge execution. If balance is insufficient, the tool returns a `payment_url` where the user can recharge. --- ## Required Environment Variables | Variable | Description | |----------|-------------| | `FEISHU_USER_ID` | Feishu user open_id for billing identification | | `OPENAI_API_KEY` | AI model API key (OpenAI, MiniMax, or OpenAI-compatible endpoint) | | `OPENAI_API_BASE` | Base URL for AI API (optional, defaults to MiniMax endpoint) | | `SKILL_BILLING_API_KEY` | Builder API Key from skillpay.me (required for paid calls) | | `SKILL_BILLING_SKILL_ID` | Skill slug on SkillPay (defaults to `data-cleaner-ai`) | --- ## Error Handling | Error | Handling | |-------|----------| | Balance insufficient | Return `payment_url` for recharge | | Network error on billing | Allow call through in dev mode (no charge) | | Tier feature not available | Skip feature gracefully, continue with available features | | No data source provided | Raise error requesting input | --- ## License MIT FILE:scripts/classifier.py """ F4 · Data classification and tagging. AI-powered auto-tagging + user-defined classification rules. Usage: from classifier import DataClassifier, load_rules, DEFAULT_RULES clf = DataClassifier(field_info, rules=rules) df_tagged, tag_col = clf.classify(df) """ import os import json import re import math from typing import Dict, List, Optional, Callable, Any from dataclasses import dataclass, field import pandas as pd from field_identifier import FieldType, FieldInfo # ─── Built-in classification rules ───────────────────────────────────────────── DEFAULT_RULES: List[Dict[str, Any]] = [ # ── Customer value tiers ──────────────────────────────────────────────────── { "name": "高价值客户", "description": "累计消费 ≥ 5000 元", "conditions": [ { "column": ".*金额|.*消费|.*总额|.*销售额", "type": "amount", "operator": "gte", "value": 5000, } ], }, { "name": "中等价值客户", "description": "累计消费 1000–5000 元", "conditions": [ { "column": ".*金额|.*消费|.*总额|.*销售额", "type": "amount", "operator": "between", "value": [1000, 5000], } ], }, { "name": "低价值客户", "description": "累计消费 < 1000 元", "conditions": [ { "column": ".*金额|.*消费|.*总额|.*销售额", "type": "amount", "operator": "lt", "value": 1000, } ], }, { "name": "沉睡用户", "description": "超过 90 天未消费", "conditions": [ { "column": ".*日期|.*时间|.*最后.*购买", "type": "days_ago", "operator": "gt", "value": 90, } ], }, { "name": "新客户", "description": "注册/首次购买在 30 天内", "conditions": [ { "column": ".*注册|.*首次|.*创建.*时间", "type": "days_ago", "operator": "lte", "value": 30, } ], }, { "name": "高风险订单", "description": "金额异常高（> 平均值 3σ）或收货地址模糊", "conditions": [ { "column": ".*金额|.*总额|.*实付", "type": "amount_outlier", "operator": "gt_3sigma", } ], }, { "name": "企业客户", "description": "邮箱为企业域名（非 gmail/qq/163 等公共邮箱）", "conditions": [ { "column": ".*邮箱|.*邮件", "type": "email_domain", "operator": "not_public", } ], }, { "name": "VIP客户", "description": "累计消费 ≥ 20000 元", "conditions": [ { "column": ".*金额|.*消费|.*总额", "type": "amount", "operator": "gte", "value": 20000, } ], }, ] # Public email domains to exclude for "企业客户" rule PUBLIC_EMAIL_DOMAINS = { "gmail.com", "qq.com", "163.com", "126.com", "sina.com", "sohu.com", "hotmail.com", "outlook.com", "yahoo.com", "foxmail.com", "google.com", "live.com", "msn.com", } # ─── Rule evaluation engine ──────────────────────────────────────────────────── @dataclass class TagResult: tag: str description: str rows_matched: int @dataclass class ClassificationReport: total_rows: int tagged_rows: int tags: List[TagResult] def summary(self) -> str: lines = [ f"总行数：{self.total_rows} | 已打标签：{self.tagged_rows}" f"（{self.tagged_rows/self.total_rows:.0%}）" if self.total_rows else "总行数：0", "标签分布：", ] for t in self.tags: lines.append(f" - {t.tag}：{t.rows_matched} 条") return "\n".join(lines) class DataClassifier: """ Apply classification rules to a DataFrame to produce tags. Parameters ---------- field_info : Dict[col -> FieldInfo] from field_identifier rules : list of rule dicts (see DEFAULT_RULES format) tag_col : name for the output tag column """ def __init__( self, field_info: Dict[str, FieldInfo], rules: Optional[List[Dict[str, Any]]] = None, tag_col: str = "标签", *, use_ai: bool = False, ai_api_key: Optional[str] = None, ): self.field_info = field_info self.rules = rules or DEFAULT_RULES self.tag_col = tag_col self.use_ai = use_ai self.ai_api_key = ai_api_key def classify( self, df: pd.DataFrame, ) -> tuple[pd.DataFrame, str, ClassificationReport]: """ Apply rules and return tagged DataFrame. Adds a column named self.tag_col. """ df = df.copy() df[self.tag_col] = "" # initialise tag_counts: Dict[str, int] = {} matched_any = 0 for rule in self.rules: mask = self._evaluate_rule(df, rule) n_matched = mask.sum() if n_matched > 0: # Append tag (may already have one) new_tags = df.loc[mask, self.tag_col].apply( lambda x: f"{x}; {rule['name']}" if x else rule["name"] ) df[self.tag_col] = new_tags.combine_first(df[self.tag_col]) tag_counts[rule["name"]] = int(n_matched) matched_any += n_matched # Remove leading semicolon on first tag df[self.tag_col] = df[self.tag_col].str.lstrip("; ") report = ClassificationReport( total_rows=len(df), tagged_rows=int((df[self.tag_col] != "").sum()), tags=[ TagResult(tag=name, description="", rows_matched=count) for name, count in tag_counts.items() ], ) return df, self.tag_col, report # ─── Rule evaluation ─────────────────────────────────────────────────────── def _evaluate_rule(self, df: pd.DataFrame, rule: Dict) -> pd.Series: """Return a boolean mask for rows matching the rule.""" import numpy as np conditions = rule.get("conditions", []) if not conditions: return pd.Series(False, index=df.index) masks: List[pd.Series] = [] for cond in conditions: col_pat = cond["column"] cond_type = cond["type"] operator = cond["operator"] value = cond.get("value") # Find matching columns matching_cols = [ c for c in df.columns if re.match(col_pat, str(c), re.IGNORECASE) ] col_mask = pd.Series(False, index=df.index) for col in matching_cols: col_mask |= self._eval_condition(df[col], cond_type, operator, value) masks.append(col_mask) # All conditions must match (AND) if not masks: return pd.Series(False, index=df.index) result = masks[0] for m in masks[1:]: result = result & m return result def _eval_condition( self, series: pd.Series, cond_type: str, operator: str, value: Any, ) -> pd.Series: """Evaluate a single condition on a series.""" import numpy as np if cond_type == "amount": # Parse numeric from string like "¥1,234.56" nums = series.astype(str).str.replace( r"[¥$€£,\s]", "", regex=True ).str.replace(r"元", "", regex=False) nums = pd.to_numeric(nums, errors="coerce").fillna(0) if operator == "gte": return nums >= float(value) elif operator == "gt": return nums > float(value) elif operator == "lt": return nums < float(value) elif operator == "lte": return nums <= float(value) elif operator == "between": lo, hi = float(value[0]), float(value[1]) return (nums >= lo) & (nums <= hi) elif cond_type == "days_ago": # Parse date and compute days since dates = series.astype(str).str.strip() now_ts = pd.Timestamp.now().timestamp() def to_days_ago(v: str): for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y年%m月%d日", "%Y%m%d", "%m/%d/%Y"): try: dt = pd.to_datetime(v, format=fmt) return (now_ts - dt.timestamp()) / 86400 except Exception: continue return -1 # can't parse days = dates.apply(to_days_ago) if operator == "gt": return days > float(value) elif operator == "lte": return (days >= 0) & (days <= float(value)) elif cond_type == "amount_outlier": if operator == "gt_3sigma": nums = pd.to_numeric( series.astype(str).str.replace(r"[¥$€£,\s]", "", regex=True), errors="coerce" ) mean = nums.mean() std = nums.std() if std == 0 or math.isnan(std): return pd.Series(False, index=series.index) return nums > (mean + 3 * std) elif cond_type == "email_domain": if operator == "not_public": domain_col = series.astype(str).str.split("@").str[-1].str.lower() return ~domain_col.isin(PUBLIC_EMAIL_DOMAINS) return pd.Series(False, index=series.index) # ─── AI-powered auto-tagging ─────────────────────────────────────────────── def classify_with_ai( self, df: pd.DataFrame, ) -> tuple[pd.DataFrame, str, ClassificationReport]: """ Use AI to generate tags when built-in rules are insufficient. Requires DATA_CLEANER_API_KEY env var. """ if not self.use_ai: return self.classify(df) api_key = self.ai_api_key or os.environ.get("DATA_CLEANER_API_KEY", "") if not api_key: # Fallback to rules return self.classify(df) # Build sample for AI samples = df.head(20).to_csv(index=False, encoding="utf-8") prompt = ( "你是一个数据分析师。根据以下数据样本，为每行生成合适的业务标签。\n" "标签例如：高价值客户、低价值客户、沉睡用户、新客户、VIP客户、" "企业客户、高风险订单、待激活用户、忠诚客户等。\n" "每行输出一个标签（最多2个，用分号分隔）。\n" "只输出CSV格式，第一列是原数据索引，第二列是标签。\n" f"\n样本数据：\n{samples}" ) try: import urllib.request url = "https://api.minimax.chat/v1/text/chatcompletion_pro" payload = json.dumps({ "model": "MiniMax-Text-01", "messages": [{"role": "user", "content": prompt}], "max_tokens": 500, "temperature": 0.3, }).encode("utf-8") req = urllib.request.Request( url, data=payload, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=20) as resp: raw = json.loads(resp.read().decode("utf-8")) content = raw["choices"][0]["message"]["content"] # Parse AI output into tags df_tagged, tag_col, report = self.classify(df) # Overlay AI tags from content df_tagged, report = self._apply_ai_tags(df_tagged, content, report) return df_tagged, tag_col, report except Exception: return self.classify(df) def _apply_ai_tags( self, df: pd.DataFrame, ai_output: str, report: ClassificationReport, ) -> tuple[pd.DataFrame, ClassificationReport]: """Parse AI output CSV and merge tags.""" import io as _io import numpy as np try: ai_df = pd.read_csv( _io.StringIO(ai_output), header=None, names=["index", "tag"], dtype={"index": str}, ) idx_map = dict(zip(ai_df["index"].astype(str), ai_df["tag"].astype(str))) for idx_str, tag in idx_map.items(): if idx_str.isdigit(): idx = int(idx_str) if idx < len(df): cur = df.at[idx, self.tag_col] or "" df.at[idx, self.tag_col] = f"{cur}; {tag}".lstrip("; ") except Exception: pass # If AI output is unparseable, keep rule-based tags report.tagged_rows = int((df[self.tag_col] != "").sum()) return df, report # ─── User rule management ───────────────────────────────────────────────────── def load_rules(path: str) -> List[Dict[str, Any]]: """Load custom classification rules from a JSON file.""" with open(path, encoding="utf-8") as f: return json.load(f) def save_rules(rules: List[Dict[str, Any]], path: str) -> None: """Save custom classification rules to a JSON file.""" with open(path, "w", encoding="utf-8") as f: json.dump(rules, f, ensure_ascii=False, indent=2) FILE:scripts/billing.py #!/usr/bin/env python3 """ Billing integration for data-cleaner-ai via SkillPay.me. Pay-per-call: $0.01 USDT per execution. Balance insufficient → payment_url returned (user tops up at skillpay.me/{slug}). Required environment variables: SKILL_BILLING_API_KEY - SkillPay Builder API Key SKILL_BILLING_SKILL_ID - SkillPay Skill ID (slug: data-cleaner-ai) FEISHU_USER_ID - User open_id for billing Billing API docs: https://skillpay.me/api/v1/billing """ import os import time import requests from pathlib import Path # ─── Constants ──────────────────────────────────────────────────────────────── BILLING_API_URL = "https://skillpay.me/api/v1/billing" CALL_PRICE = 0.0100 # USDT per execution # Cache TTL: 5 minutes _CACHE_TTL = 300 # ─── Internal cache ───────────────────────────────────────────────────────────── _cache: dict = {} def _cache_get(key: str) -> dict | None: entry = _cache.get(key) if entry is None: return None if time.time() - entry["_ts"] > _CACHE_TTL: del _cache[key] return None return entry def _cache_set(key: str, data: dict) -> None: _cache[key] = {**data, "_ts": time.time()} # ─── Headers ─────────────────────────────────────────────────────────────────── def _get_headers() -> dict: return { "X-API-Key": os.environ.get("SKILL_BILLING_API_KEY", ""), "Content-Type": "application/json", } def _get_skill_id() -> str: return os.environ.get("SKILL_BILLING_SKILL_ID", "data-cleaner-ai") # ─── Dev mode (no API key configured) ───────────────────────────────────────── def _is_dev_mode() -> bool: key = os.environ.get("SKILL_BILLING_API_KEY", "").strip() return key == "" # ─── Check balance ──────────────────────────────────────────────────────────── def check_balance(user_id: str) -> dict: """ Returns current user balance in USDT. Returns: {"balance": float, "ok": bool} """ if _is_dev_mode(): return {"balance": 999.0, "ok": True} cache_key = f"balance_{user_id}" cached = _cache_get(cache_key) if cached: return {"balance": cached["balance"], "ok": True} try: resp = requests.get( f"{BILLING_API_URL}/balance", headers=_get_headers(), params={"user_id": user_id, "skill_id": _get_skill_id()}, timeout=10, ) data = resp.json() balance = float(data.get("balance", 0.0)) _cache_set(cache_key, {"balance": balance}) return {"balance": balance, "ok": True} except Exception: return {"balance": 999.0, "ok": True} # Fail open: allow on network error # ─── Charge user ─────────────────────────────────────────────────────────────── def charge_user(user_id: str) -> dict: """ Charge user for one execution ($0.01 USDT). Returns: {"ok": True, "balance": float} on success {"ok": False, "balance": float, "payment_url": str} on insufficient balance """ if _is_dev_mode(): return {"ok": True, "balance": 999.0} try: resp = requests.post( f"{BILLING_API_URL}/charge", headers=_get_headers(), json={ "user_id": user_id, "skill_id": _get_skill_id(), "amount": CALL_PRICE, }, timeout=10, ) data = resp.json() if data.get("success"): return { "ok": True, "balance": float(data.get("balance", 0.0)), } return { "ok": False, "balance": float(data.get("balance", 0.0)), "payment_url": data.get("payment_url", ""), } except Exception: # Network error → allow in dev mode return {"ok": True, "balance": 999.0} # ─── Payment link ────────────────────────────────────────────────────────────── def get_payment_link(user_id: str) -> str: """ Get payment URL for user to top up. """ if _is_dev_mode(): return "" try: resp = requests.post( f"{BILLING_API_URL}/payment-link", headers=_get_headers(), json={"user_id": user_id, "skill_id": _get_skill_id()}, timeout=10, ) data = resp.json() return data.get("payment_url", "") except Exception: return "" FILE:scripts/reporter.py """ F6 · Data quality report generation. Generates structured reports in Markdown (for Feishu Doc) and dict form. Usage: from reporter import DataQualityReporter, Report reporter = DataQualityReporter(before_df, after_df, field_info) report = reporter.generate() markdown = reporter.to_markdown(report) """ import math from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from datetime import datetime import pandas as pd from field_identifier import FieldType, FieldInfo, FIELD_TYPE_LABELS from cleaner import CleaningReport as CleaningReport_ from classifier import ClassificationReport as ClassificationReport_ # ─── Report dataclasses ───────────────────────────────────────────────────────── @dataclass class ColumnStats: col: str field_type: str total: int missing: int missing_pct: float unique: int sample_values: List[str] = field(default_factory=list) @dataclass class DataQualityReport: generated_at: str source_name: str tier: str before_shape: tuple[int, int] after_shape: tuple[int, int] duplicate_rate_before: float duplicate_rate_after: float missing_rate_before: float missing_rate_after: float overall_score: float # 0-100 column_stats: List[ColumnStats] = field(default_factory=list) cleaning_report: Optional[CleaningReport_] = None classification_report: Optional[ClassificationReport_] = None recommendations: List[str] = field(default_factory=list) # ─── Reporter ────────────────────────────────────────────────────────────────── class DataQualityReporter: """ Generate data quality reports before/after cleaning. Parameters ---------- before_df : original DataFrame after_df : cleaned DataFrame field_info : Dict[col -> FieldInfo] source_name: str identifier for the data source tier : subscription tier string cleaning_report : Optional[CleaningReport] classification_report: Optional[ClassificationReport] """ def __init__( self, before_df: pd.DataFrame, after_df: pd.DataFrame, field_info: Dict[str, FieldInfo], *, source_name: str = "数据源", tier: str = "free", cleaning_report: Optional[CleaningReport_] = None, classification_report: Optional[ClassificationReport_] = None, ): self.before_df = before_df.copy() self.after_df = after_df.copy() self.field_info = field_info self.source_name = source_name self.tier = tier self.cleaning_report = cleaning_report self.classification_report = classification_report def generate(self) -> DataQualityReport: """Build the full quality report.""" stats = self._compute_stats() dup_before = self._dup_rate(self.before_df) dup_after = self._dup_rate(self.after_df) miss_before = self._missing_rate(self.before_df) miss_after = self._missing_rate(self.after_df) score = self._overall_score(dup_after, miss_after) recs = self._recommendations(stats, dup_after, miss_after) return DataQualityReport( generated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), source_name=self.source_name, tier=self.tier, before_shape=(len(self.before_df), len(self.before_df.columns)), after_shape =(len(self.after_df), len(self.after_df.columns)), duplicate_rate_before=dup_before, duplicate_rate_after=dup_after, missing_rate_before=miss_before, missing_rate_after=miss_after, overall_score=score, column_stats=stats, cleaning_report=self.cleaning_report, classification_report=self.classification_report, recommendations=recs, ) # ─── Per-column statistics ───────────────────────────────────────────────── def _compute_stats(self) -> List[ColumnStats]: stats = [] for col in self.before_df.columns: fi = self.field_info.get(col) ft_label = FIELD_TYPE_LABELS.get(fi.field_type, "未知") if fi else "未知" total = len(self.before_df) missing = self.before_df[col].astype(str).isin( ["", "nan", "NaN", "None", "null", "NULL", "undefined", "未知"] ).sum() missing_pct = missing / total if total else 0 unique = self.before_df[col].nunique() samples = ( self.before_df[col] .astype(str) .loc[lambda x: ~x.isin(["", "nan", "NaN", "None", "null", "未知"])] .drop_duplicates() .head(5) .tolist() ) stats.append(ColumnStats( col=str(col), field_type=ft_label, total=total, missing=missing, missing_pct=missing_pct, unique=unique, sample_values=samples, )) return stats # ─── Rate calculations ─────────────────────────────────────────────────────── def _dup_rate(self, df: pd.DataFrame) -> float: if df.empty: return 0.0 return 1.0 - (df.drop_duplicates().shape[0] / len(df)) def _missing_rate(self, df: pd.DataFrame) -> float: if df.empty: return 0.0 total_cells = df.shape[0] * df.shape[1] missing_cells = df.astype(str).isin( ["", "nan", "NaN", "None", "null", "NULL", "undefined", "未知"] ).sum().sum() return missing_cells / total_cells if total_cells else 0.0 def _overall_score(self, dup_rate: float, miss_rate: float) -> float: """Score 0-100, higher = better quality.""" dup_penalty = dup_rate * 40 # up to -40 miss_penalty = miss_rate * 40 # up to -40 return round(max(0.0, 100.0 - (dup_rate * 40) - (miss_rate * 40)), 1) # ─── Recommendations ──────────────────────────────────────────────────────── def _recommendations( self, stats: List[ColumnStats], dup_rate: float, miss_rate: float, ) -> List[str]: recs: List[str] = [] if dup_rate > 0.1: recs.append( f"⚠️ 重复率较高（{dup_rate:.1%}），建议检查是否存在系统重复导入。" ) if miss_rate > 0.05: recs.append( f"⚠️ 缺失率偏高（{miss_rate:.1%}），建议补充缺失数据以提高分析准确性。" ) for s in stats: if s.missing_pct > 0.3: recs.append( f"⚠️ 列「{s.col}」缺失率达 {s.missing_pct:.0%}，" f"建议确认该字段是否必填或启用智能补全功能。" ) if s.unique == 1 and s.total > 10: recs.append( f"ℹ️ 列「{s.col}」所有值相同（{s.unique} 个唯一值），" f"可能为无效特征，建议移除。" ) if not recs: recs.append("✅ 数据质量良好，未发现明显问题。") return recs # ─── Markdown output ──────────────────────────────────────────────────────── def to_markdown(self, report: DataQualityReport) -> str: """Render report as Feishu-compatible Markdown.""" lines: List[str] = [] lines.append(f"# 📊 数据质量报告") lines.append(f"") lines.append(f"**数据源：** {report.source_name}") lines.append(f"**版本：** {report.tier}") lines.append(f"**生成时间：** {report.generated_at}") lines.append("") # Score card score_emoji = ( "🟢" if report.overall_score >= 80 else "🟡" if report.overall_score >= 60 else "🔴" ) lines.append(f"## {score_emoji} 综合质量评分：{report.overall_score}/100") lines.append("") # Shape summary lines.append("## 📐 数据规模") lines.append("") lines.append("| 阶段 | 行数 | 列数 |") lines.append("|------|------|------|") lines.append( f"| 清洗前 | {report.before_shape[0]:,} | {report.before_shape[1]:,} |" ) lines.append( f"| 清洗后 | {report.after_shape[0]:,} | {report.after_shape[1]:,} |" ) lines.append("") # Key rates lines.append("## 📈 质量指标") lines.append("") lines.append("| 指标 | 清洗前 | 清洗后 | 变化 |") lines.append("|------|--------|--------|------|") dup_delta = report.duplicate_rate_after - report.duplicate_rate_before dup_arrow = "↓" if dup_delta < 0 else "↑" lines.append( f"| 重复率 | {report.duplicate_rate_before:.2%} " f"| {report.duplicate_rate_after:.2%} | {dup_arrow} {abs(dup_delta):.2%} |" ) miss_delta = report.missing_rate_after - report.missing_rate_before miss_arrow = "↓" if miss_delta < 0 else "↑" lines.append( f"| 缺失率 | {report.missing_rate_before:.2%} " f"| {report.missing_rate_after:.2%} | {miss_arrow} {abs(miss_delta):.2%} |" ) lines.append("") # Cleaning details if report.cleaning_report: cr = report.cleaning_report lines.append("## 🧹 清洗详情") lines.append("") lines.append(f"- 原始行数：{cr.original_rows}") lines.append(f"- 清洗后行数：{cr.cleaned_rows}") lines.append(f"- 去重：移除 {cr.duplicates_removed} 条") lines.append(f"- 补全：填补 {cr.missing_filled} 个缺失值") lines.append(f"- 格式化：处理 {cr.formatted_cells} 个单元格") lines.append("") if cr.missing_by_column: lines.append("**各列缺失情况：**") lines.append("") for col, n in cr.missing_by_column.items(): lines.append(f"- {col}：{n} 个缺失值") lines.append("") if cr.duplicate_groups: lines.append(f"检测到 {len(cr.duplicate_groups)} 组重复记录，已自动去重。") lines.append("") # Classification tags if report.classification_report: cl = report.classification_report lines.append("## 🏷️ 标签分布") lines.append("") lines.append(f"总行数：{cl.total_rows} | 已打标签：{cl.tagged_rows}") lines.append("") for t in cl.tags: pct = t.rows_matched / cl.total_rows if cl.total_rows else 0 lines.append(f"- **{t.tag}**：{t.rows_matched} 条（{pct:.0%}）") lines.append("") # Per-column stats lines.append("## 📋 字段质量详情") lines.append("") lines.append( "| 字段名 | 类型 | 样本量 | 缺失 | 缺失率 | 唯一值 | 示例值 |" ) lines.append( "|--------|------|--------|------|--------|--------|--------|" ) for s in report.column_stats: sample_str = " / ".join(s.sample_values[:2]) if len(sample_str) > 30: sample_str = sample_str[:27] + "..." lines.append( f"| {s.col} | {s.field_type} | {s.total} | {s.missing} " f"| {s.missing_pct:.0%} | {s.unique} | {sample_str} |" ) lines.append("") # Recommendations if report.recommendations: lines.append("## 💡 优化建议") lines.append("") for rec in report.recommendations: lines.append(f"{rec}") lines.append("") return "\n".join(lines) # ─── Compact dict (for JSON / API response) ───────────────────────────────── def to_dict(self, report: DataQualityReport) -> Dict[str, Any]: return { "generated_at": report.generated_at, "source_name": report.source_name, "tier": report.tier, "before_shape": report.before_shape, "after_shape": report.after_shape, "duplicate_rate_before": report.duplicate_rate_before, "duplicate_rate_after": report.duplicate_rate_after, "missing_rate_before": report.missing_rate_before, "missing_rate_after": report.missing_rate_after, "overall_score": report.overall_score, "column_stats": [ { "col": s.col, "field_type": s.field_type, "total": s.total, "missing": s.missing, "missing_pct": round(s.missing_pct, 4), "unique": s.unique, "sample_values": s.sample_values, } for s in report.column_stats ], "recommendations": report.recommendations, } FILE:scripts/__init__.py """Multi-Source Data Cleanser - Skill scripts package.""" __version__ = "1.0.0" FILE:scripts/field_identifier.py """ F2 · Intelligent field type identification. Uses regex patterns + optional AI (MiniMax / DeepSeek) for ambiguous fields. Supported types: name, phone, email, address, amount, date, sku, order_no, id_card, gender, url, ip_address, bank_account, custom Usage: from field_identifier import identify_fields, FieldType types = identify_fields(df) # dict: col_name -> FieldType types = identify_fields(df, ai_model="deepseek") # use AI for uncertain cols types = identify_fields(df, custom_rules={"姓名": "name"}) """ import os import re import json import math from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, field from enum import Enum import pandas as pd # ─── Field type enum ─────────────────────────────────────────────────────────── class FieldType(str, Enum): NAME = "name" PHONE = "phone" EMAIL = "email" ADDRESS = "address" AMOUNT = "amount" DATE = "date" SKU = "sku" ORDER_NO = "order_no" ID_CARD = "id_card" GENDER = "gender" URL = "url" IP_ADDRESS = "ip_address" BANK_ACCOUNT = "bank_account" TEXT = "text" NUMBER = "number" UNKNOWN = "unknown" FIELD_TYPE_LABELS: Dict[FieldType, str] = { FieldType.NAME: "姓名", FieldType.PHONE: "手机号", FieldType.EMAIL: "邮箱", FieldType.ADDRESS: "地址", FieldType.AMOUNT: "金额", FieldType.DATE: "日期", FieldType.SKU: "SKU", FieldType.ORDER_NO: "订单号", FieldType.ID_CARD: "身份证", FieldType.GENDER: "性别", FieldType.URL: "网址", FieldType.IP_ADDRESS: "IP地址", FieldType.BANK_ACCOUNT: "银行账号", FieldType.TEXT: "文本", FieldType.NUMBER: "数字", FieldType.UNKNOWN: "未知", } # ─── Regex patterns ──────────────────────────────────────────────────────────── _PATTERNS: Dict[FieldType, re.Pattern] = { FieldType.EMAIL: re.compile( r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$" ), FieldType.PHONE: re.compile( r"^1[3-9]\d[\s\-]?\d{4}[\s\-]?\d{4}$" # Chinese mobile ), FieldType.ID_CARD: re.compile( r"^\d{15}$|^\d{17}[\dXx]$" ), FieldType.DATE: re.compile( r"^(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}[日]?)" r"|(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})" r"|^\d{8}$" r"|^\d{10}$|^\d{13}$" ), # AMOUNT: must have decimal point or currency symbol (not pure digits) FieldType.AMOUNT: re.compile( r"^[¥$€£]\s*[\d，,]+\.\d+$" r"|^[¥$€£]\s*[\d，,]+\$" r"|^[\d，,]+\.\d+\s*[¥$€£]?$" ), # SKU: letter prefix or specific formats, NOT bare 11-digit phone-like numbers FieldType.SKU: re.compile( r"^[A-Za-z]+[\-]?\d{4,10}$" r"|^SKU[\-_]?\d{4,12}$" r"|^[A-Z0-9]{6,12}$" ), FieldType.ORDER_NO: re.compile( r"^(DD|ORDER|PO|SO|BM)[\-_]?\d{4,16}$" r"|^[A-Z]{2,}\d{6,16}$" r"|^\d{16,24}$" ), FieldType.URL: re.compile( r"^https?://[^\s]+$" ), FieldType.IP_ADDRESS: re.compile( r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$" ), FieldType.GENDER: re.compile( r"^(男|女|男性|女性|M|m|F|f)$" ), } # Chinese province / city / district keywords for address detection _ADDRESS_KEYWORDS = [ "省", "市", "区", "县", "镇", "乡", "村", "路", "街", "号", "栋", "单元", "室", "弄", "巷", "楼", "广场", "大厦", "小区", "北京", "上海", "深圳", "广州", "杭州", "成都", "武汉", "西安", ] _GENDER_VALUES = {"男", "女", "男性", "女性", "M", "F", "m", "f", "0", "1"} # ─── Scoring helpers ─────────────────────────────────────────────────────────── def _is_empty(val: str) -> bool: return val in ("", "nan", "NaN", "None", "null", "NULL", "undefined") def _match_score(pat: re.Pattern, val: str) -> float: """Fraction of non-empty values matching the pattern (0.0–1.0).""" if _is_empty(val): return 0.0 return 1.0 if pat.match(str(val).strip()) else 0.0 def _avg_match_score(series: pd.Series, pat: re.Pattern) -> float: vals = [str(v) for v in series if not _is_empty(str(v))] if not vals: return 0.0 return sum(1 for v in vals if pat.match(v.strip())) / len(vals) def _numeric_score(series: pd.Series) -> float: vals = [str(v).strip() for v in series if not _is_empty(str(v))] if not vals: return 0.0 count = 0 for v in vals: # Remove currency symbols and commas cleaned = re.sub(r"[¥$€£,\s]", "", v) try: float(cleaned) count += 1 except ValueError: pass return count / len(vals) def _date_score(series: pd.Series) -> float: """Higher score if values look like dates.""" vals = [str(v).strip() for v in series if not _is_empty(str(v))] if not vals: return 0.0 date_forms = [ (r"\d{4}[-/年]\d{1,2}[-/月]\d{1,2}", 0.9), (r"\d{8}", 0.8), (r"\d{10}", 0.8), (r"\d{1,2}[-/]\d{1,2}[-/]\d{2,4}", 0.7), ] total = 0.0 for v in vals: for pat, w in date_forms: if re.match(pat, v): total += w break return total / len(vals) def _address_score(series: pd.Series) -> float: vals = [str(v) for v in series if not _is_empty(str(v))] if not vals: return 0.0 hits = sum(1 for v in vals if any(k in v for k in _ADDRESS_KEYWORDS)) avg_len = sum(len(v) for v in vals) / len(vals) # Longish strings with address keywords → high score len_bonus = min(avg_len / 20, 1.0) * 0.3 return hits / len(vals) * 0.7 + len_bonus def _gender_score(series: pd.Series) -> float: vals = {str(v).strip() for v in series if not _is_empty(str(v))} if not vals: return 0.0 intersection = vals & _GENDER_VALUES return min(len(intersection) / 2, 1.0) # just 1-2 distinct values matters def _name_score(series: pd.Series) -> float: """ Detect Chinese / common personal names. Chinese names: 2-4 chars, often with common surname characters, no digits, no special chars beyond hyphen/·. """ # Common Chinese surname characters (top 100) COMMON_SURNAMES = set( "李王张刘陈杨赵黄周吴徐孙胡朱高林何郭马罗梁宋郑谢韩唐冯于董萧程曹袁邓许傅沈曾彭吕苏卢蒋蔡贾丁魏薛叶阎余潘杜戴夏钟汪田任姜范方石姚谭廖邹熊金陆郝孔白崔康毛邱秦江史顾侯邵孟龙万段雷钱汤尹黎易常武乔贺赖龚文" ) vals = [str(v).strip() for v in series if not _is_empty(str(v))] if not vals: return 0.0 hits = 0 for v in vals: # Length check: 2-4 chars typical for Chinese full name if 2 <= len(v) <= 4: # No digits if re.match(r"^[\u4e00-\u9fa5·\-']+$", v): # All Chinese + hyphen/apostrophe # Check for surname character at start if v[0] in COMMON_SURNAMES: hits += 1 elif re.match(r"^[A-Z][a-z]+$", v): # English name "John Smith" hits += 0.8 elif re.match(r"^[A-Za-z]+$", v) and 3 <= len(v) <= 15: # Single-word English name hits += 0.5 # English name formats elif re.match(r"^[A-Z][a-z]+(\s+[A-Z][a-z]+)+$", v): hits += 1 return hits / len(vals) # ─── Column name heuristics ──────────────────────────────────────────────────── _COL_NAME_WEIGHTS: Dict[str, Dict[FieldType, float]] = { # Name → FieldType hints "name": {FieldType.NAME: 0.9}, "姓名": {FieldType.NAME: 0.9}, "客户姓名": {FieldType.NAME: 0.9}, "phone": {FieldType.PHONE: 0.9}, "tel": {FieldType.PHONE: 0.8}, "mobile": {FieldType.PHONE: 0.9}, "电话": {FieldType.PHONE: 0.9}, "手机": {FieldType.PHONE: 0.9}, "手机号": {FieldType.PHONE: 0.9}, "email": {FieldType.EMAIL: 0.9}, "mail": {FieldType.EMAIL: 0.8}, "邮箱": {FieldType.EMAIL: 0.9}, "电子邮件": {FieldType.EMAIL: 0.9}, "address": {FieldType.ADDRESS: 0.9}, "地址": {FieldType.ADDRESS: 0.9}, "收货地址": {FieldType.ADDRESS: 0.9}, "amount": {FieldType.AMOUNT: 0.9}, "金额": {FieldType.AMOUNT: 0.9}, "总价": {FieldType.AMOUNT: 0.9}, "单价": {FieldType.AMOUNT: 0.9}, "price": {FieldType.AMOUNT: 0.9}, "销售额": {FieldType.AMOUNT: 0.9}, "date": {FieldType.DATE: 0.9}, "日期": {FieldType.DATE: 0.9}, "成交日期": {FieldType.DATE: 0.9}, "订单日期": {FieldType.DATE: 0.9}, "下单时间": {FieldType.DATE: 0.9}, "创建时间": {FieldType.DATE: 0.9}, "sku": {FieldType.SKU: 0.9}, "SKU": {FieldType.SKU: 0.9}, "商品编号": {FieldType.SKU: 0.9}, "order": {FieldType.ORDER_NO: 0.9}, "订单": {FieldType.ORDER_NO: 0.9}, "订单号": {FieldType.ORDER_NO: 0.9}, "order_no": {FieldType.ORDER_NO: 0.9}, "订单编号": {FieldType.ORDER_NO: 0.9}, "id_card": {FieldType.ID_CARD: 0.9}, "身份证": {FieldType.ID_CARD: 0.9}, "gender": {FieldType.GENDER: 0.9}, "性别": {FieldType.GENDER: 0.9}, "url": {FieldType.URL: 0.9}, "网址": {FieldType.URL: 0.9}, "ip": {FieldType.IP_ADDRESS: 0.9}, "ip地址": {FieldType.IP_ADDRESS: 0.9}, "银行账号": {FieldType.BANK_ACCOUNT: 0.9}, "账号": {FieldType.BANK_ACCOUNT: 0.5}, } def _col_name_score(col: str, ftype: FieldType) -> float: col_lower = col.lower() hints = _COL_NAME_WEIGHTS.get(col, {}) hints.update(_COL_NAME_WEIGHTS.get(col_lower, {})) return hints.get(ftype, 0.0) # ─── Main identification function ───────────────────────────────────────────── @dataclass class FieldInfo: field_type: FieldType confidence: float # 0.0–1.0 samples: List[str] = field(default_factory=list) def label(self) -> str: return FIELD_TYPE_LABELS.get(self.field_type, str(self.field_type)) def identify_fields( df: pd.DataFrame, *, custom_rules: Optional[Dict[str, str]] = None, ai_model: Optional[str] = None, ai_api_key: Optional[str] = None, ) -> Dict[str, FieldInfo]: """ Identify field types for all columns in a DataFrame. Parameters ---------- df : input DataFrame custom_rules : {column_name: FieldType.value} user-defined overrides ai_model : "minimax" or "deepseek" — use AI for uncertain columns ai_api_key : API key (defaults to env DATA_CLEANER_API_KEY) Returns ------- Dict[col_name -> FieldInfo] """ custom_rules = custom_rules or {} results: Dict[str, FieldInfo] = {} uncertain_cols: List[str] = [] for col in df.columns: col_str = str(col).strip() # 1. User override if col_str in custom_rules: try: ft = FieldType(custom_rules[col_str]) results[col] = FieldInfo(ft, 1.0, []) continue except ValueError: pass # 2. Compute pattern scores scores: Dict[FieldType, float] = {} series = df[col].astype(str) for ftype, pat in _PATTERNS.items(): pattern_score = _avg_match_score(series, pat) name_score = _col_name_score(col_str, ftype) # Combine: pattern is primary, name hints are bonus combined = pattern_score * 0.7 + name_score * 0.3 if combined > 0: scores[ftype] = min(combined, 1.0) # Special fast-path scorers — add to scores only if the best # type-specific pattern score is low (NUMBER is a generic fallback, # not a competitor to clear type signals like phone/email/ID). num_score = _numeric_score(series) date_s = _date_score(series) addr_s = _address_score(series) gender_s = _gender_score(series) name_s = _name_score(series) # Chinese / common name pattern best_type_score = max(scores.values()) if scores else 0.0 # NUMBER only wins when no specific type has scored clearly (below 0.7) if num_score > 0.5 and best_type_score < 0.7: scores[FieldType.NUMBER] = num_score if date_s > best_type_score: scores[FieldType.DATE] = date_s if addr_s > 0.5: scores[FieldType.ADDRESS] = addr_s if gender_s > 0.5: scores[FieldType.GENDER] = gender_s if name_s > best_type_score: scores[FieldType.NAME] = name_s # 3. Pick best match if not scores: results[col] = FieldInfo(FieldType.UNKNOWN, 0.0, _samples(series, 3)) uncertain_cols.append(col) continue best_ftype = max(scores, key=lambda k: scores[k]) confidence = scores[best_ftype] # Low confidence → mark uncertain if confidence < 0.6: uncertain_cols.append(col) results[col] = FieldInfo( field_type=best_ftype, confidence=confidence, samples=_samples(series, 3), ) # 4. AI assist for uncertain columns if uncertain_cols and ai_model: _fill_with_ai(results, uncertain_cols, df, ai_model, ai_api_key) return results def _samples(series: pd.Series, n: int = 3) -> List[str]: vals = [str(v) for v in series if str(v) not in ("", "nan", "NaN")] return list(dict.fromkeys(vals))[:n] # unique, preserve order # ─── AI-assisted identification ─────────────────────────────────────────────── def _fill_with_ai( results: Dict[str, FieldInfo], uncertain_cols: List[str], df: pd.DataFrame, model: str, api_key: Optional[str], ) -> None: api_key = api_key or os.environ.get("DATA_CLEANER_API_KEY", "") if not api_key: return import urllib.request import urllib.parse col_samples = { col: _samples(df[col].astype(str), 5) for col in uncertain_cols } prompt = ( "你是一个数据分析师。请根据以下列名和样本值，判断每列的字段类型。\n" "字段类型可选：姓名、手机号、邮箱、地址、金额、日期、SKU、订单号、身份证、性别、网址、IP地址、银行账号、文本、数字、未知\n" "只输出JSON对象，格式：{\"列名\": \"类型\"}\n" f"\n列名与样本值：\n{json.dumps(col_samples, ensure_ascii=False)}\n" ) try: if model in ("deepseek", "minimax"): # Unified MiniMax/DeepSeek compatible API url = "https://api.minimax.chat/v1/text/chatcompletion_pro" if model == "deepseek": url = "https://api.deepseek.com/v1/chat/completions" payload = json.dumps({ "model": "MiniMax-Text-01" if model == "minimax" else "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "max_tokens": 300, "temperature": 0.1, }).encode("utf-8") req = urllib.request.Request( url, data=payload, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, method="POST", ) with urllib.request.urlopen(req, timeout=15) as resp: raw = json.loads(resp.read().decode("utf-8")) content = raw["choices"][0]["message"]["content"] # Parse JSON from response match = re.search(r"\{[^{}]*\}", content, re.DOTALL) if match: ai_result = json.loads(match.group()) for col, type_str in ai_result.items(): if col in results: try: ft = FieldType(type_str) results[col] = FieldInfo(ft, 0.8, results[col].samples) except ValueError: pass except Exception: pass # AI fallback: keep regex result # ─── User-defined field mapping ─────────────────────────────────────────────── def apply_custom_mapping( field_info: Dict[str, FieldInfo], mapping: Dict[str, str], ) -> Dict[str, FieldInfo]: """ Apply user-defined column → field_type overrides. mapping = {"姓名": "name", "电话": "phone"} """ for col, type_str in mapping.items(): if col in field_info: try: ft = FieldType(type_str) field_info[col] = FieldInfo(ft, 1.0, field_info[col].samples) except ValueError: pass return field_info # ─── Summary ────────────────────────────────────────────────────────────────── def field_summary(field_info: Dict[str, FieldInfo]) -> str: lines = ["### 字段识别结果"] for col, info in field_info.items(): conf_pct = f"{info.confidence:.0%}" lines.append(f"- **{col}** → {info.label()}（置信度 {conf_pct}）") return "\n".join(lines) FILE:scripts/parser.py """ F1 · Multi-format parser Supports: Excel (.xlsx/.xls), CSV/TSV, JSON, clipboard-pasted text. Usage: from parser import parse_file, parse_text df = parse_file("path/to/data.xlsx") df = parse_file("path/to/data.csv") df = parse_text("姓名,电话\\n张三,13800138000") """ import io import json import re import pandas as pd from pathlib import Path from typing import Union, List, Optional # ─── Helpers ───────────────────────────────────────────────────────────────── def _sniff_delimiter(text: str) -> str: """Sniff the most likely delimiter from the first few lines.""" sample = text[:2000] for delim in [",", "\t", "|", ";"]: if delim in sample: return delim return "," def _is_json(text: str) -> bool: text = text.strip() return text.startswith("{") or text.startswith("[") def _is_excel(path: Union[str, Path]) -> bool: p = str(path).lower() return p.endswith(".xlsx") or p.endswith(".xls") # ─── Core parsers ───────────────────────────────────────────────────────────── def parse_file( path: Union[str, Path], sheet: Optional[Union[str, int]] = None, encoding: str = "utf-8", ) -> pd.DataFrame: """ Auto-detect format and parse a file into a DataFrame. Parameters ---------- path : file path sheet : for Excel files, sheet name or 0-indexed position encoding : text file encoding Returns ------- pd.DataFrame """ path = str(path) if _is_excel(path): return _parse_excel(path, sheet=sheet) suffix = Path(path).suffix.lower() if suffix in (".csv", ".tsv", ".txt"): return _parse_csv(path, delimiter=None, encoding=encoding) if suffix == ".json": return _parse_json(path) # fallback: try CSV then JSON try: return _parse_csv(path, delimiter=None, encoding=encoding) except Exception: return _parse_json(path) def parse_text(text: str, format_hint: str = "auto") -> pd.DataFrame: """ Parse clipboard / pasted text into a DataFrame. Parameters ---------- text : raw pasted content format_hint : "csv" | "tsv" | "json" | "auto" Returns ------- pd.DataFrame """ text = text.strip() if not text: raise ValueError("粘贴内容为空，请确认已复制数据。") if format_hint == "json" or (format_hint == "auto" and _is_json(text)): return _parse_json_str(text) if format_hint == "tsv" or (format_hint == "auto" and "\t" in text): return _parse_csv_str(text, delimiter="\t") # Default: sniff delimiter delim = _sniff_delimiter(text) if format_hint == "auto" else "," return _parse_csv_str(text, delimiter=delim) # ─── Private implementations ─────────────────────────────────────────────────── def _parse_excel(path: str, sheet: Optional[Union[str, int]]) -> pd.DataFrame: try: import openpyxl except ImportError: raise ImportError("openpyxl is required to read .xlsx files. Install: pip install openpyxl") try: import xlrd except ImportError: pass # .xls may not be readable but .xlsx will work xl_kwargs: dict = {"engine": "openpyxl"} if Path(path).suffix.lower() == ".xls": xl_kwargs["engine"] = "xlrd" if sheet is not None: xl_kwargs["sheet_name"] = sheet else: xl_kwargs["sheet_name"] = 0 # first sheet by default return pd.read_excel(path, **xl_kwargs) def _parse_csv(path: str, delimiter: Optional[str], encoding: str) -> pd.DataFrame: if delimiter: return pd.read_csv(path, delimiter=delimiter, encoding=encoding, dtype=str, keep_default_na=False) # Auto-sniff try: return pd.read_csv(path, encoding=encoding, dtype=str, keep_default_na=False) except UnicodeDecodeError: for enc in ["gbk", "gb2312", "latin1"]: try: return pd.read_csv(path, encoding=enc, dtype=str, keep_default_na=False) except UnicodeDecodeError: continue raise ValueError(f"无法解析 CSV 文件 {path}，请确认文件编码（UTF-8 / GBK）。") def _parse_json(path: str) -> pd.DataFrame: with open(path, encoding="utf-8") as f: data = json.load(f) return _json_to_df(data) def _parse_csv_str(text: str, delimiter: str = ",") -> pd.DataFrame: df = pd.read_csv(io.StringIO(text), delimiter=delimiter, dtype=str, keep_default_na=False) # Strip whitespace from all cells df = df.apply(lambda col: col.astype(str).str.strip()) return df def _parse_json_str(text: str) -> pd.DataFrame: data = json.loads(text) return _json_to_df(data) def _json_to_df(data) -> pd.DataFrame: """ Handle three common JSON shapes: 1. Array of flat objects : [{a:1},{a:2}] 2. Object with array field: {rows:[{a:1},...]} 3. Nested / hierarchical : flatten if possible """ if isinstance(data, list): records = data elif isinstance(data, dict): # Try to find the array field for key in ["data", "rows", "records", "items", "list", "result"]: if key in data and isinstance(data[key], list): records = data[key] break else: records = [data] else: raise ValueError("JSON 结构不支持，请提供对象数组或包含数据数组的根对象。") if not records: return pd.DataFrame() df = pd.DataFrame(records) # Flatten any deeply nested columns (one level only) for col in df.columns: if df[col].apply(lambda x: isinstance(x, dict)).any(): exploded = df[col].apply(lambda x: x if isinstance(x, dict) else {}) norm = pd.json_normalize(exploded) norm.columns = [f"{col}_{k}" for k in norm.columns] df = pd.concat([df.drop(columns=[col]), norm], axis=1) df = df.apply(lambda col: col.astype(str).str.strip()) return df # ─── Multi-file loader ───────────────────────────────────────────────────────── def load_sources( sources: List[Union[str, Path, io.IOBase]], texts: Optional[List[str]] = None, ) -> List[pd.DataFrame]: """ Load multiple data sources (files or pasted texts) into DataFrames. Returns list of (name, df) tuples. """ results = [] # Files for src in (sources or []): name = Path(src).name if isinstance(src, (str, Path)) else str(src) try: df = parse_file(src) results.append((name, df)) except Exception as exc: raise ValueError(f"文件「{name}」解析失败：{exc}") # Pasted texts for i, text in enumerate(texts or []): name = f"粘贴数据_{i+1}" try: df = parse_text(text) results.append((name, df)) except Exception as exc: raise ValueError(f"「{name}」解析失败：{exc}") return results # ─── Quick peek (header preview) ────────────────────────────────────────────── def preview_file(path: Union[str, Path], n: int = 5) -> pd.DataFrame: """Return first n rows without full load (for display).""" df = parse_file(path) return df.head(n) def preview_text(text: str, n: int = 5) -> pd.DataFrame: """Return first n rows of pasted text (for display).""" df = parse_text(text) return df.head(n) FILE:scripts/cleaner.py """ F3 · Core data cleaning engine. Handles: - Smart deduplication (exact + fuzzy) - Missing value imputation (mean/mode/inference/leave_blank) - Format unification (phone/date/amount/address) Usage: from cleaner import DataCleaner cleaner = DataCleaner(field_info) df_clean = cleaner.clean(df) """ import re import math from typing import Dict, Optional, List, Callable, Any from dataclasses import dataclass, field from datetime import datetime import pandas as pd from field_identifier import FieldType, FieldInfo # ─── Address standardisation helpers ───────────────────────────────────────── _CN_PROVINCES = [ "北京市","天津市","上海市","重庆市", "河北省","山西省","辽宁省","吉林省","黑龙江省", "江苏省","浙江省","安徽省","福建省","江西省","山东省", "河南省","湖北省","湖南省","广东省","海南省", "四川省","贵州省","云南省","陕西省","甘肃省","青海省","台湾省", "内蒙古自治区","广西壮族自治区","西藏自治区","宁夏回族自治区","新疆维吾尔自治区", "香港特别行政区","澳门特别行政区", ] _CN_PROVINCE_ABBR = { "北京":"北京市","上海":"上海市","天津":"天津市","重庆":"重庆市", "河北":"河北省","山西":"山西省","辽宁":"辽宁省","吉林":"吉林省", "黑龙江":"黑龙江省","江苏":"江苏省","浙江":"浙江省","安徽":"安徽省", "福建":"福建省","江西":"江西省","山东":"山东省","河南":"河南省", "湖北":"湖北省","湖南":"湖南省","广东":"广东省","海南":"海南省", "四川":"四川省","贵州":"贵州省","云南":"云南省","陕西":"陕西省", "甘肃":"甘肃省","青海":"青海省","台湾":"台湾省", "内蒙古":"内蒙古自治区","广西":"广西壮族自治区","西藏":"西藏自治区", "宁夏":"宁夏回族自治区","新疆":"新疆维吾尔自治区", "香港":"香港特别行政区","澳门":"澳门特别行政区", } # ─── Dataclass results ───────────────────────────────────────────────────────── @dataclass class CleaningReport: original_rows: int cleaned_rows: int duplicates_removed: int missing_filled: int formatted_cells: int missing_by_column: Dict[str, int] = field(default_factory=dict) duplicate_groups: List[List[int]] = field(default_factory=list) def summary(self) -> str: return ( f"原始行数：{self.original_rows} | 清洗后行数：{self.cleaned_rows}\n" f"去重：移除 {self.duplicates_removed} 条重复记录\n" f"补全：填补 {self.missing_filled} 个缺失值\n" f"格式化：处理 {self.formatted_cells} 个单元格" ) # ─── Core Cleaner ───────────────────────────────────────────────────────────── class DataCleaner: """ Main cleaning orchestrator. Parameters ---------- field_info : Dict[col -> FieldInfo] from field_identifier """ def __init__( self, field_info: Dict[str, FieldInfo], *, dedup_strategy: str = "auto", fill_strategy: str = "auto", format_phone: bool = True, format_date: bool = True, format_amount: bool = True, format_address: bool = True, ): self.field_info = field_info self.dedup_strategy = dedup_strategy # "exact" | "fuzzy" | "auto" self.fill_strategy = fill_strategy # "auto" | "mean" | "mode" | "leave_blank" self.format_phone = format_phone self.format_date = format_date self.format_amount = format_amount self.format_address = format_address # Per-column type lookup self._type_map: Dict[str, FieldType] = { col: fi.field_type for col, fi in field_info.items() } def clean(self, df: pd.DataFrame) -> tuple[pd.DataFrame, CleaningReport]: """ Full cleaning pipeline on a single DataFrame. Returns (cleaned_df, report). """ df = df.copy() original_rows = len(df) total_missing_filled = 0 total_formatted = 0 # ── Step 1: Normalise column names ────────────────────────────────────── df.columns = [str(c).strip() for c in df.columns] # ── Step 2: Missing value imputation ───────────────────────────────────── df, missing_filled, missing_by_col = self._impute(df) total_missing_filled += missing_filled # ── Step 3: Format unification ────────────────────────────────────────── df, formatted = self._format_all(df) total_formatted += formatted # ── Step 4: Deduplication ─────────────────────────────────────────────── df, dup_removed, dup_groups = self._deduplicate(df) total_dup_removed = original_rows - len(df) report = CleaningReport( original_rows=original_rows, cleaned_rows=len(df), duplicates_removed=total_dup_removed, missing_filled=total_missing_filled, formatted_cells=total_formatted, missing_by_column=missing_by_col, duplicate_groups=dup_groups, ) return df, report # ── Imputation ───────────────────────────────────────────────────────────── def _impute( self, df: pd.DataFrame, ) -> tuple[pd.DataFrame, int, Dict[str, int]]: """Fill missing values according to field type and fill_strategy.""" filled = 0 missing_by_col: Dict[str, int] = {} df = df.copy() for col in df.columns: ftype = self._type_map.get(col, FieldType.UNKNOWN) blanks = df[col].astype(str).isin(["", "nan", "NaN", "None", "null", "NULL", "undefined"]) n_blank = blanks.sum() if n_blank == 0: continue missing_by_col[col] = int(n_blank) strategy = self.fill_strategy if strategy == "leave_blank": continue # keep NaN filled_col = self._fill_column(df[col], ftype, strategy) df[col] = filled_col filled += n_blank return df, filled, missing_by_col def _fill_column( self, series: pd.Series, ftype: FieldType, strategy: str, ) -> pd.Series: """Apply the appropriate fill logic to a single column.""" blanks = series.astype(str).isin(["", "nan", "NaN", "None", "null", "NULL"]) if strategy == "mean": if ftype in (FieldType.AMOUNT, FieldType.NUMBER): nums = pd.to_numeric(series, errors="coerce") mean_val = nums.mean() if not math.isnan(mean_val): filled = series.copy() filled[blanks] = f"{mean_val:.2f}" return filled elif strategy == "mode": # Most common non-blank value non_blanks = series[~blanks] if not non_blanks.empty: mode_val = non_blanks.mode() if not mode_val.empty: filled = series.copy() filled[blanks] = mode_val.iloc[0] return filled elif strategy in ("auto", "inference"): filled = self._auto_fill(series, ftype) return filled # Default: leave blank return series def _auto_fill(self, series: pd.Series, ftype: FieldType) -> pd.Series: """ Auto-fill missing values based on field semantics. Returns a new series (does not modify in place). """ blanks = series.astype(str).isin(["", "nan", "NaN", "None", "null", "NULL"]) if blanks.sum() == 0: return series filled = series.copy() if ftype == FieldType.GENDER: # Fill with most common non_blank = series[~blanks] if not non_blank.empty: mode = non_blank.mode() if not mode.empty: filled[blanks] = mode.iloc[0] elif ftype in (FieldType.AMOUNT, FieldType.NUMBER): nums = pd.to_numeric(series, errors="coerce") mean_val = nums.mean() if not math.isnan(mean_val): filled[blanks] = f"{mean_val:.2f}" elif ftype == FieldType.DATE: # Try to parse most common format, fill with placeholder filled[blanks] = "未知" elif ftype == FieldType.PHONE: filled[blanks] = "未知" elif ftype == FieldType.EMAIL: filled[blanks] = "未知" # Text / address / others → "未知" elif ftype in (FieldType.TEXT, FieldType.ADDRESS, FieldType.UNKNOWN, FieldType.NAME, FieldType.SKU, FieldType.ORDER_NO): filled[blanks] = "未知" return filled # ── Format unification ─────────────────────────────────────────────────────── def _format_all(self, df: pd.DataFrame) -> tuple[pd.DataFrame, int]: formatted = 0 df = df.copy() for col in df.columns: ftype = self._type_map.get(col, FieldType.UNKNOWN) before = df[col].astype(str) if ftype == FieldType.PHONE and self.format_phone: df[col] = df[col].apply(self._format_phone) elif ftype == FieldType.DATE and self.format_date: df[col] = df[col].apply(self._format_date) elif ftype == FieldType.AMOUNT and self.format_amount: df[col] = df[col].apply(self._format_amount_) elif ftype == FieldType.ADDRESS and self.format_address: df[col] = df[col].apply(self._standardise_address) after = df[col].astype(str) formatted += (before != after).sum() return df, formatted def _format_phone(self, val: Any) -> str: """Normalise phone to 1xx-xxxx-xxxx.""" s = str(val).strip() if s in ("", "nan", "NaN", "None", "未知"): return s # Strip all non-digit digits = re.sub(r"\D", "", s) # Take last 11 digits if > 11 if len(digits) > 11: digits = digits[-11:] if len(digits) == 11 and digits[0] == "1": return f"{digits[0:3]}-{digits[3:7]}-{digits[7:11]}" return s # return original if not recognisable def _format_date(self, val: Any) -> str: """Normalise date to YYYY-MM-DD.""" s = str(val).strip() if s in ("", "nan", "NaN", "None", "未知"): return s # Already ISO? if re.match(r"\d{4}-\d{2}-\d{2}", s): return s # Chinese format: YYYY年MM月DD日 m = re.match(r"(\d{4})年(\d{1,2})月(\d{1,2})日?", s) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" # Slash/ hyphen: YYYY/MM/DD or YYYY-MM-DD or YYYYMD for pat in [ r"(\d{4})[-/年](\d{1,2})[-/月](\d{1,2})", r"(\d{4})(\d{2})(\d{2})", ]: m = re.match(pat, s) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" # YYYYMMDD integer if re.match(r"^\d{8}$", s): return f"{s[:4]}-{s[4:6]}-{s[6:8]}" # Unix timestamp (seconds) if re.match(r"^\d{10}$", s): try: from datetime import datetime as dt return dt.fromtimestamp(int(s)).strftime("%Y-%m-%d") except Exception: pass # Milliseconds if re.match(r"^\d{13}$", s): try: from datetime import datetime as dt return dt.fromtimestamp(int(s) / 1000).strftime("%Y-%m-%d") except Exception: pass return s def _format_amount_(self, val: Any) -> str: """Normalise amount to two decimal places.""" s = str(val).strip() if s in ("", "nan", "NaN", "None", "未知"): return s # Strip currency symbols and commas cleaned = re.sub(r"[¥$€£,\s]", "", s) try: num = float(cleaned) return f"{num:.2f}" except ValueError: return s def _standardise_address(self, val: Any) -> str: """Standardise Chinese address to 省市区街道格式.""" s = str(val).strip() if s in ("", "nan", "NaN", "None", "未知"): return s # Expand province abbreviation for abbr, full in _CN_PROVINCE_ABBR.items(): if s.startswith(abbr): s = full + s[len(abbr):] break # Normalise separators to " " s = re.sub(r"[,，;；\t]+", " ", s) s = re.sub(r"\s+", " ", s).strip() return s # ─── Deduplication ───────────────────────────────────────────────────────── def _deduplicate( self, df: pd.DataFrame, ) -> tuple[pd.DataFrame, int, List[List[int]]]: """ Deduplicate rows. Strategy: exact → drop_duplicates on all columns fuzzy → fuzzy match on key identity columns (phone/name/email/order_no) auto → fuzzy if key columns present, else exact """ strategy = self.dedup_strategy dup_groups: List[List[int]] = [] if df.empty: return df, 0, dup_groups key_cols = self._find_key_columns() original_len = len(df) if strategy == "exact": before = len(df) df = df.drop_duplicates() removed = before - len(df) return df, removed, dup_groups if strategy in ("fuzzy", "auto") and key_cols: df, removed, dup_groups = self._fuzzy_dedup(df, key_cols) return df, removed, dup_groups # Fallback: exact before = len(df) df = df.drop_duplicates() return df, before - len(df), dup_groups def _find_key_columns(self) -> List[str]: """Find identity-like columns suitable for fuzzy dedup.""" key_types = { FieldType.PHONE, FieldType.EMAIL, FieldType.NAME, FieldType.ORDER_NO, FieldType.SKU, FieldType.ID_CARD, } return [ col for col, ft in self._type_map.items() if ft in key_types ] def _fuzzy_dedup( self, df: pd.DataFrame, key_cols: List[str], ) -> tuple[pd.DataFrame, int, List[List[int]]]: """ Fuzzy deduplication using FuzzyWuzzy on key columns. Keeps the first occurrence, removes fuzzy duplicates. """ try: from fuzzywuzzy import fuzz except ImportError: # Fallback to exact if fuzzywuzzy not installed before = len(df) df = df.drop_duplicates(subset=key_cols, keep="first") return df, before - len(df), [] # Build composite key strings key_strs = df[key_cols].astype(str).agg(" | ".join, axis=1) keep_idx: List[int] = [] dup_groups: List[List[int]] = [] removed = 0 for i, key in enumerate(key_strs): is_dup = False for j in keep_idx: score = fuzz.ratio(key, key_strs.iloc[j]) if score >= 88: # threshold is_dup = True break if is_dup: removed += 1 else: keep_idx.append(i) dup_indices = set(range(len(df))) - set(keep_idx) dup_groups = [list(dup_indices)] if dup_indices else [] return df.iloc[keep_idx].reset_index(drop=True), removed, dup_groups # ─── Convenience function ───────────────────────────────────────────────────── def clean_dataframe( df: pd.DataFrame, field_info: Dict[str, FieldInfo], *, dedup_strategy: str = "auto", fill_strategy: str = "auto", ) -> tuple[pd.DataFrame, CleaningReport]: """ One-liner clean. """ cleaner = DataCleaner( field_info, dedup_strategy=dedup_strategy, fill_strategy=fill_strategy, ) return cleaner.clean(df) FILE:scripts/tier_limits.py """ Tier-based access control for Multi-Source Data Cleanser. Tiers: FREE - 50 rows/month, 1 source, single format, no merge/AI/report BASIC - 500 rows/month, 3 sources, basic dedup STD - 3000 rows/month, unlimited sources, format unification + smart fill PRO - unlimited, multi-source merge + AI classification + data quality report Usage: from tier_limits import check_tier, TIER_LIMITS, TierLimitExceeded check_tier("free", rows=30) # raises if over limit check_tier("basic", sources=2) # raises if over limit has_feature("pro", "fuzzy_join") # bool enforce_max_rows(tier, rows, count_key) # updates monthly counter """ import os import json import time from pathlib import Path import time from enum import Enum from dataclasses import dataclass, field from typing import Optional, Dict, Any # ─── Constants ──────────────────────────────────────────────────────────────── class Tier(str, Enum): FREE = "free" BASIC = "basic" STD = "std" PRO = "pro" TIER_ORDER = [Tier.FREE, Tier.BASIC, Tier.STD, Tier.PRO] # Monthly row limits per tier TIER_MONTHLY_ROWS: Dict[Tier, int] = { Tier.FREE: 50, Tier.BASIC: 500, Tier.STD: 3000, Tier.PRO: -1, # unlimited } # Max data sources (files / pasted blocks) TIER_MAX_SOURCES: Dict[Tier, int] = { Tier.FREE: 1, Tier.BASIC: 3, Tier.STD: -1, Tier.PRO: -1, } # Max columns allowed TIER_MAX_COLUMNS: Dict[Tier, int] = { Tier.FREE: 10, Tier.BASIC: 50, Tier.STD: 200, Tier.PRO: -1, } # Feature gates per tier TIER_FEATURES: Dict[Tier, set] = { Tier.FREE: { "single_format", # CSV or XLSX only "basic_dedup", # exact match dedup }, Tier.BASIC: { "single_format", "basic_dedup", "multi_format", # CSV + XLSX + TSV }, Tier.STD: { "single_format", "basic_dedup", "multi_format", "smart_fill", # mean/mode/inference imputation "format_unification", # phone/date/amount standardization "advanced_dedup", # fuzzy dedup }, Tier.PRO: { "single_format", "basic_dedup", "multi_format", "smart_fill", "format_unification", "advanced_dedup", "fuzzy_join", # multi-source merge "ai_classification", # AI tagging "data_quality_report", # quality report → Feishu doc "bitable_output", # write to Feishu Bitable "unlimited_rows", }, } # ─── State file ──────────────────────────────────────────────────────────────── def _get_state_path() -> str: """Resolve state file path dynamically (env var may be set after import).""" return os.environ.get( "DATA_CLEANER_STATE_FILE", "/tmp/data_cleaner_state.json" ) def _load_state() -> Dict[str, Any]: try: with open(_get_state_path()) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} def _save_state(state: Dict[str, Any]) -> None: with open(_get_state_path(), "w") as f: json.dump(state, f, ensure_ascii=False) def _get_month_key() -> str: """YYYY-MM string for monthly reset.""" return time.strftime("%Y-%m") # ─── Exceptions ───────────────────────────────────────────────────────────────── class TierLimitExceeded(Exception): """Raised when user exceeds their tier quota.""" def __init__(self, message: str, tier: Tier, limit_name: str): super().__init__(message) self.tier = tier self.limit_name = limit_name class FeatureNotAvailable(Exception): """Raised when tier does not support a feature.""" def __init__(self, feature: str, tier: Tier): super().__init__( f"功能「{feature}」仅在标准版/专业版可用。" f"当前版本：{tier.value}。" f"请升级以解锁此功能。" ) self.feature = feature self.tier = tier # ─── Core API ────────────────────────────────────────────────────────────────── # ─── Token Verification ──────────────────────────────────────────────────────── # For ClawHub SkillPay version: tier is inferred from API_KEY prefix only. # No network call to yk-global.com — removed. VALID_PREFIXES = { "GEO", "PROFIT", "INV", "DATA", "MON", "PDF", "BANK", "CONTRACT", "EMAIL", "CONV", "RPT", "SENTIMENT", } # Cache TTL: 5 minutes _CACHE_TTL = 300 def _prefix_to_tier(api_key: str) -> Tier: """Infer Tier from key prefix (no network call).""" if not api_key: return Tier.FREE upper = api_key.upper() if "ENT" in upper: return Tier.PRO if "MAX" in upper: return Tier.PRO if "PRO" in upper: return Tier.PRO if "STD" in upper: return Tier.STD if "BSC" in upper: return Tier.BASIC if "FREE" in upper: return Tier.FREE return Tier.FREE def _is_valid_key_format(api_key: str) -> bool: """Check if key matches 91Skillhub prefix pattern.""" if not api_key or "-" not in api_key: return False prefix = api_key.split("-")[0].upper() return prefix in VALID_PREFIXES # ─── Core API ────────────────────────────────────────────────────────────────── def get_user_tier() -> Tier: """ Resolve current user's subscription tier. ClawHub version: prefix-based inference only (no network call). Fallback: DATA_CLEANER_TIER env var → FREE. """ api_key = os.environ.get("DATA_CLEANER_API_KEY", "") if api_key and _is_valid_key_format(api_key): return _prefix_to_tier(api_key) # Fallback: manual tier override raw = os.environ.get("DATA_CLEANER_TIER", Tier.FREE.value).lower() try: return Tier(raw) except ValueError: return Tier.FREE def has_feature(tier: Tier, feature: str) -> bool: """Check if a tier supports a named feature.""" return feature in TIER_FEATURES.get(tier, set()) def check_feature(tier: Tier, feature: str) -> None: """Raise FeatureNotAvailable if feature is not available for tier.""" if not has_feature(tier, feature): raise FeatureNotAvailable(feature, tier) def check_tier( tier: Tier, *, rows: Optional[int] = None, sources: Optional[int] = None, columns: Optional[int] = None, ) -> None: """ Validate that the requested operation fits within tier limits. Raises TierLimitExceeded if not. """ state = _load_state() month = _get_month_key() # ── Monthly rows ── if rows is not None: limit = TIER_MONTHLY_ROWS[tier] used = state.get("usage", {}).get(month, {}).get("rows", 0) if limit > 0 and (used + rows) > limit: raise TierLimitExceeded( f"本月已使用 {used} 条，剩余 {limit - used} 条。" f"免费版每月限额 50 条。", tier, "monthly_rows", ) # ── Data sources ── if sources is not None: limit = TIER_MAX_SOURCES[tier] if limit > 0 and sources > limit: raise TierLimitExceeded( f"免费版最多支持 1 个数据源，基础版支持 3 个。" f"当前操作涉及 {sources} 个数据源。", tier, "max_sources", ) # ── Columns ── if columns is not None: limit = TIER_MAX_COLUMNS[tier] if limit > 0 and columns > limit: raise TierLimitExceeded( f"免费版最多支持 10 列，当前数据有 {columns} 列。", tier, "max_columns", ) def record_usage(*, rows: int = 0) -> Dict[str, Any]: """ Increment monthly usage counters. Returns updated state. Call after each successful清洗 operation. """ state = _load_state() month = _get_month_key() usage = state.setdefault("usage", {}) month_data = usage.setdefault(month, {"rows": 0}) month_data["rows"] = month_data.get("rows", 0) + rows state["usage"][month] = month_data _save_state(state) return month_data def get_usage_summary(tier: Tier) -> Dict[str, Any]: """Return current month usage stats for display.""" state = _load_state() month = _get_month_key() month_data = state.get("usage", {}).get(month, {"rows": 0}) limit = TIER_MONTHLY_ROWS[tier] return { "month": month, "rows_used": month_data.get("rows", 0), "rows_limit": limit if limit > 0 else "unlimited", "tier": tier.value, } def tier_display_name(tier: Tier) -> str: names = { Tier.FREE: "免费版", Tier.BASIC: "基础版", Tier.STD: "标准版", Tier.PRO: "专业版", } return names.get(tier, tier.value) FILE:scripts/main.py #!/usr/bin/env python3 """ Multi-Source Data Cleanser — main orchestrator. Usage (CLI): python main.py clean --input data.xlsx --output cleaned.xlsx python main.py merge --sources data1.csv data2.csv --on "手机号" --output merged.csv python main.py report --input cleaned.xlsx --output report.md Usage (import): from main import run_clean_pipeline result = run_clean_pipeline("data.xlsx") """ import os import sys import json import argparse import tempfile from pathlib import Path from typing import Dict, List, Optional, Any import pandas as pd # ─── Import skill modules ─────────────────────────────────────────────────────── from parser import parse_file, parse_text, load_sources from field_identifier import identify_fields, FieldType, FieldInfo from cleaner import DataCleaner, clean_dataframe, CleaningReport from classifier import DataClassifier, ClassificationReport from merger import DataMerger, MergeResult from reporter import DataQualityReporter from output import DataExporter, ExportError from tier_limits import ( Tier, get_user_tier, check_feature, check_tier, record_usage, get_usage_summary, tier_display_name, FeatureNotAvailable, TierLimitExceeded, ) from billing import charge_user # ─── Tier-based feature availability check ──────────────────────────────────── def _resolve_tier(tier_name: Optional[str]) -> Tier: if tier_name: try: return Tier(tier_name.lower()) except ValueError: pass return get_user_tier() # ─── Main pipeline ───────────────────────────────────────────────────────────── def run_clean_pipeline( sources: Optional[List[str]] = None, texts: Optional[List[str]] = None, *, tier: Optional[str] = None, output_format: str = "xlsx", output_path: Optional[str] = None, custom_field_mapping: Optional[Dict[str, str]] = None, dedup_strategy: str = "auto", fill_strategy: str = "auto", classify: bool = False, ai_model: Optional[str] = None, generate_report: bool = True, bitable_output: bool = False, feishu_open_id: Optional[str] = None, feishu_folder_token: Optional[str] = None, report_title: str = "数据质量报告", ) -> Dict[str, Any]: """ Full cleaning pipeline. Parameters ---------- sources : list of file paths texts : list of pasted text blocks tier : subscription tier override (free/basic/std/pro) output_format : "xlsx" | "csv" output_path : output file path custom_field_mapping : {col_name: field_type} overrides dedup_strategy : "exact" | "fuzzy" | "auto" fill_strategy : "auto" | "mean" | "mode" | "leave_blank" classify : whether to run AI classification ai_model : "minimax" or "deepseek" for AI features generate_report : whether to generate quality report bitable_output : write to Feishu Bitable (PRO only) feishu_open_id : user's Feishu open_id feishu_folder_token : Feishu folder token for doc output report_title : title for the quality report document Returns ------- Dict with keys: cleaned_df, report_md, report_dict, file_path, bitable_result, doc_result, tier, usage_summary """ t = _resolve_tier(tier) # ── Billing (SkillPay per-call) ────────────────────────────────────────── user_id = feishu_open_id or os.environ.get("FEISHU_USER_ID", "") if user_id: bill = charge_user(user_id) if not bill["ok"]: return { "ok": False, "error": "Insufficient balance", "balance": bill["balance"], "payment_url": bill.get("payment_url", ""), "tier": tier_display_name(t), } result: Dict[str, Any] = {"tier": tier_display_name(t), "ok": True} # ── Load sources ──────────────────────────────────────────────────────────── src_list = sources or [] if not src_list and not (texts or []): raise ValueError("请提供至少一个数据源（文件路径或粘贴文本）。") # Tier: check max sources n_sources = len(src_list) + len(texts or []) check_tier(t, sources=n_sources) raw_sources = load_sources(src_list, texts) if len(raw_sources) == 1: raw_name, raw_df = raw_sources[0] else: # Multiple sources → merge first merger = DataMerger(raw_sources) try: merge_result = merger.merge(how="outer") except Exception: # Fallback: concatenate raw_df = pd.concat([df for _, df in raw_sources], ignore_index=True) raw_name = "+".join([n for n, _ in raw_sources]) merge_result = None else: raw_df = merge_result.df raw_name = f"{merge_result.left_name}+{merge_result.right_name}" # Tier: check columns check_tier(t, columns=len(raw_df.columns)) # ── Identify fields ──────────────────────────────────────────────────────── field_info = identify_fields( raw_df, custom_rules=custom_field_mapping, ai_model=ai_model, ) # ── Clean ────────────────────────────────────────────────────────────────── check_tier(t, rows=len(raw_df)) df_clean, clean_report = clean_dataframe( raw_df, field_info, dedup_strategy=dedup_strategy, fill_strategy=fill_strategy, ) # ── Classify (PRO only) ──────────────────────────────────────────────────── class_report: Optional[ClassificationReport] = None if classify: try: check_feature(t, "ai_classification") except FeatureNotAvailable: pass # Silently skip if tier doesn't support else: clf = DataClassifier(field_info, use_ai=(ai_model is not None)) df_clean, _, class_report = clf.classify_with_ai(df_clean) # ── Export cleaned data ─────────────────────────────────────────────────── exporter = DataExporter(df_clean, field_info, tier=t.value) if output_format == "csv": out_path = output_path or tempfile.mktemp(suffix=".csv") file_path = exporter.to_csv(out_path) result["file_path"] = file_path else: out_path = output_path or tempfile.mktemp(suffix=".xlsx") file_path = exporter.to_excel(out_path) result["file_path"] = file_path result["cleaned_rows"] = len(df_clean) result["cleaned_columns"] = len(df_clean.columns) # ── Record usage ────────────────────────────────────────────────────────── usage = record_usage(rows=len(raw_df)) result["usage"] = get_usage_summary(t) # ── Bitable output (PRO only) ───────────────────────────────────────────── bitable_result: Optional[Dict] = None if bitable_output: try: check_feature(t, "bitable_output") bitable_result = exporter.to_bitable( table_name="清洗结果", folder_token=feishu_folder_token, open_id=feishu_open_id, ) result["bitable"] = bitable_result except FeatureNotAvailable: result["bitable_error"] = "飞书多维表格导出仅限专业版使用。" except ExportError as e: result["bitable_error"] = str(e) # ── Quality report ──────────────────────────────────────────────────────── report_md: Optional[str] = None doc_result: Optional[Dict] = None if generate_report: try: check_feature(t, "data_quality_report") except FeatureNotAvailable: # Still generate dict report reporter = DataQualityReporter( raw_df, df_clean, field_info, source_name=raw_name, tier=tier_display_name(t), cleaning_report=clean_report, classification_report=class_report, ) report = reporter.generate() result["report_dict"] = reporter.to_dict(report) else: reporter = DataQualityReporter( raw_df, df_clean, field_info, source_name=raw_name, tier=tier_display_name(t), cleaning_report=clean_report, classification_report=class_report, ) report = reporter.generate() report_md = reporter.to_markdown(report) result["report_md"] = report_md result["report_dict"] = reporter.to_dict(report) # Create Feishu doc try: doc_result = exporter.to_feishu_doc( report_markdown=report_md, title=report_title, folder_token=feishu_folder_token, ) result["doc"] = doc_result except ExportError as e: result["doc_error"] = str(e) result["clean_report"] = { "original_rows": clean_report.original_rows, "cleaned_rows": clean_report.cleaned_rows, "duplicates_removed": clean_report.duplicates_removed, "missing_filled": clean_report.missing_filled, "formatted_cells": clean_report.formatted_cells, } return result # ─── Standalone merge pipeline ───────────────────────────────────────────────── def run_merge_pipeline( sources: List[str], on: Optional[List[str]] = None, fuzzy_on: Optional[List[str]] = None, *, tier: Optional[str] = None, output_format: str = "xlsx", output_path: Optional[str] = None, fuzzy_threshold: int = 85, ) -> Dict[str, Any]: """ Multi-source merge pipeline. Parameters ---------- sources : list of file paths (2 or more) on : list of column names for exact join fuzzy_on : list of column names for fuzzy join fuzzy_threshold : fuzzy match score threshold (0-100) """ t = _resolve_tier(tier) # ── Billing (SkillPay per-call) ────────────────────────────────────────── user_id = os.environ.get("FEISHU_USER_ID", "") if user_id: bill = charge_user(user_id) if not bill["ok"]: return { "ok": False, "error": "Insufficient balance", "balance": bill["balance"], "payment_url": bill.get("payment_url", ""), "tier": tier_display_name(t), } check_feature(t, "fuzzy_join") raw_sources = load_sources(sources) merger = DataMerger(raw_sources) # Build (left_col, right_col) pairs on_pairs = [] fuzzy_pairs = [] if on: for col in on: on_pairs.append((col, col)) if fuzzy_on: for col in fuzzy_on: fuzzy_pairs.append((col, col)) merge_result = merger.merge( how="outer", on=on_pairs or None, fuzzy_on=fuzzy_pairs or None, fuzzy_threshold=fuzzy_threshold, ) # Clean merged result field_info = identify_fields(merge_result.df) df_clean, clean_report = clean_dataframe(merge_result.df, field_info) # Export exporter = DataExporter(df_clean, field_info, tier=t.value) out_path = output_path or tempfile.mktemp(suffix=f".{output_format}") if output_format == "csv": file_path = exporter.to_csv(out_path) else: file_path = exporter.to_excel(out_path) usage = record_usage(rows=len(merge_result.df)) return { "file_path": file_path, "merge_summary": merge_result.summary(), "clean_report": { "original_rows": clean_report.original_rows, "cleaned_rows": clean_report.cleaned_rows, "duplicates_removed": clean_report.duplicates_removed, }, "tier": tier_display_name(t), "usage": get_usage_summary(t), } # ─── CLI entry point ─────────────────────────────────────────────────────────── def _cli(): parser = argparse.ArgumentParser( description="多源数据清洗器 - CLI", formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = parser.add_subparsers(dest="cmd", required=True) # clean c = sub.add_parser("clean", help="清洗数据") c.add_argument("--input", "-i", help="输入文件路径") c.add_argument("--output", "-o", help="输出文件路径") c.add_argument("--format", "-f", default="xlsx", choices=["xlsx", "csv"], help="输出格式（默认 xlsx）") c.add_argument("--tier", default=None) c.add_argument("--dedup", default="auto", choices=["exact", "fuzzy", "auto"]) c.add_argument("--fill", default="auto", choices=["auto", "mean", "mode", "leave_blank"]) c.add_argument("--classify", action="store_true") c.add_argument("--ai", default=None, choices=["minimax", "deepseek"]) c.add_argument("--no-report", action="store_true") c.add_argument("--text", "-t", help="粘贴文本（会覆盖 --input）") c.add_argument("--report-title", default="数据质量报告") args = parser.parse_args() kwargs = dict( tier=args.tier, output_format=args.format, output_path=args.output, dedup_strategy=args.dedup, fill_strategy=args.fill, classify=args.classify, ai_model=args.ai, generate_report=not args.no_report, report_title=args.report_title, ) if args.text: kwargs["texts"] = [args.text] elif args.input: kwargs["sources"] = [args.input] else: print("错误：必须提供 --input 或 --text", file=sys.stderr) sys.exit(1) result = run_clean_pipeline(**kwargs) print("\n✅ 清洗完成！") print(f"输出文件：{result.get('file_path')}") print(f"清洗后行数：{result.get('cleaned_rows', '?')}") print(f"版本：{result.get('tier')}") print(f"本月已用：{result.get('usage', {}).get('rows_used', '?')} 条") if "bitable" in result: print(f"飞书多维表格：{result['bitable']['url']}") if "doc" in result: print(f"质量报告文档：{result['doc']['url']}") if __name__ == "__main__": _cli() FILE:scripts/output.py """ F6 · Output module. Exports cleaned data and quality reports to: - Excel (.xlsx) - CSV - Feishu Bitable (multi-dimensional table) - Feishu Cloud Document (quality report) Usage: from output import DataExporter exporter = DataExporter(df, field_info) path = exporter.to_excel("/tmp/cleaned.xlsx") path = exporter.to_csv("/tmp/cleaned.csv") bitable_url = exporter.to_bitable(table_name="清洗结果") doc_url = exporter.to_feishu_doc(report_markdown, title="数据质量报告") """ import os import io import time import base64 import tempfile from typing import Dict, List, Optional, Any, Tuple from pathlib import Path import pandas as pd # ─── Exceptions ───────────────────────────────────────────────────────────────── class ExportError(Exception): pass # ─── Field type → Bitable field type mapping ─────────────────────────────────── LARK_FIELD_TYPES: Dict[str, int] = { "text": 1, "number": 2, "single_select": 3, "multi_select": 4, "date": 5, "checkbox": 7, "person": 11, "phone": 13, "url": 15, "attachment": 17, "created_time": 1001, "modified_time": 1002, } def _infer_bitable_field_type(field_type_str: str) -> int: """Map our FieldType label to Feishu Bitable field type ID.""" mapping: Dict[str, int] = { "姓名": 1, "手机号": 13, "邮箱": 1, "地址": 1, "金额": 2, "日期": 5, "SKU": 1, "订单号": 1, "身份证": 1, "性别": 3, "网址": 15, "IP地址": 1, "银行账号": 1, "文本": 1, "数字": 2, "未知": 1, "标签": 4, # tag column → multi-select } return mapping.get(field_type_str, 1) # ─── DataExporter ─────────────────────────────────────────────────────────────── class DataExporter: """ Export cleaned DataFrame to various formats. Parameters ---------- df : cleaned DataFrame field_info : Dict[col -> FieldInfo] for type hints tier : subscription tier (for feature gating) """ def __init__( self, df: pd.DataFrame, field_info: Optional[Dict] = None, tier: str = "free", ): self.df = df.copy() self.field_info = field_info or {} self.tier = tier # ─── Excel ───────────────────────────────────────────────────────────────── def to_excel( self, path: Optional[str] = None, sheet_name: str = "清洗结果", ) -> str: """ Write DataFrame to .xlsx file. Returns the file path. """ if path is None: path = tempfile.mktemp(suffix=".xlsx") try: import openpyxl except ImportError: raise ExportError( "openpyxl 未安装。请运行：pip install openpyxl" ) # Use xlsxwriter if available for better formatting try: import xlsxwriter # noqa: F401 self.df.to_excel(path, sheet_name=sheet_name, index=False, engine="xlsxwriter") except ImportError: self.df.to_excel(path, sheet_name=sheet_name, index=False, engine="openpyxl") return path def to_csv( self, path: Optional[str] = None, encoding: str = "utf-8-sig", ) -> str: """ Write DataFrame to CSV file. Returns the file path. """ if path is None: path = tempfile.mktemp(suffix=".csv") self.df.to_csv(path, index=False, encoding=encoding) return path def to_base64_csv(self, encoding: str = "utf-8-sig") -> str: """Return CSV as base64-encoded string (for file attachments).""" buf = io.StringIO() self.df.to_csv(buf, index=False, encoding=encoding) return base64.b64encode(buf.getvalue().encode(encoding)).decode() def to_base64_excel(self) -> str: """Return Excel as base64-encoded string.""" buf = io.BytesIO() self.df.to_excel(buf, index=False, engine="openpyxl") buf.seek(0) return base64.b64encode(buf.read()).decode() # ─── Feishu Bitable ──────────────────────────────────────────────────────── def to_bitable( self, table_name: str = "清洗结果", folder_token: Optional[str] = None, open_id: Optional[str] = None, ) -> Dict[str, Any]: """ Create a Feishu Bitable app and write data into it. Requires Feishu API credentials. Returns dict with app_token, table_id, url. Raises ExportError if Bitable creation fails. """ # Check tier if self.tier not in ("std", "pro"): raise ExportError( "飞书多维表格导出仅在标准版/专业版可用。" "请升级以解锁此功能。" ) # Lazy import feishu tools to avoid hard dependency try: from feishu_bitable_app import feishu_bitable_app from feishu_bitable_app_table import feishu_bitable_app_table from feishu_bitable_app_table_record import feishu_bitable_app_table_record except ImportError: raise ExportError( "飞书多维表格模块不可用。" "请确认已安装 feishu-bitable skill 并重启服务。" ) # 1. Create the Bitable app app_result = feishu_bitable_app( action="create", name=f"数据清洗_{table_name}", folder_token=folder_token or "", ) if not app_result.get("app_token"): raise ExportError(f"创建多维表格失败：{app_result}") app_token = app_result["app_token"] # 2. Define fields fields_def = self._build_bitable_fields() # 3. Create the table with fields table_result = feishu_bitable_app_table( action="create", app_token=app_token, table={"name": table_name, "fields": fields_def}, ) table_id = table_result.get("table_id", "") # 4. Batch write records (max 500 per call) records = self._df_to_records() BATCH = 500 for i in range(0, len(records), BATCH): batch = records[i:i + BATCH] feishu_bitable_app_table_record( action="batch_create", app_token=app_token, table_id=table_id, records=[{"fields": r} for r in batch], ) url = f"https://aiplayer.feishu.cn/bitable/{app_token}" return { "app_token": app_token, "table_id": table_id, "url": url, "rows_written": len(records), } def _build_bitable_fields(self) -> List[Dict]: """Build field definitions for Bitable table creation.""" fields = [] for col in self.df.columns: type_label = "文本" if col in self.field_info: type_label = self.field_info[col].label() if hasattr( self.field_info[col], "label" ) else str(self.field_info[col]) elif "标签" in col: type_label = "标签" # Map to bitable type ID bitable_type = _infer_bitable_field_type(type_label) field_def: Dict[str, Any] = { "field_name": str(col), "type": bitable_type, } # Single-select needs options if bitable_type == 3: unique_vals = self.df[col].dropna().unique().tolist()[:20] field_def["property"] = { "options": [ {"name": str(v)[:50]} for v in unique_vals ] } fields.append(field_def) return fields def _df_to_records(self) -> List[Dict[str, Any]]: """Convert DataFrame rows to Bitable record format.""" records = [] for _, row in self.df.iterrows(): fields: Dict[str, Any] = {} for col, val in row.items(): s = str(val) if s in ("", "nan", "NaN", "None", "未知"): fields[str(col)] = None elif "标签" in str(col): # Multi-select: split by semicolon tags = [t.strip() for t in s.split(";") if t.strip()] fields[str(col)] = tags else: fields[str(col)] = s records.append(fields) return records # ─── Feishu Cloud Document (report) ───────────────────────────────────────── def to_feishu_doc( self, report_markdown: str, title: str = "数据质量报告", folder_token: Optional[str] = None, wiki_space_id: Optional[str] = None, ) -> Dict[str, Any]: """ Create a Feishu Cloud Document with the quality report. Returns dict with doc_token, url. """ try: from feishu_create_doc import feishu_create_doc except ImportError: raise ExportError( "飞书文档模块不可用。" "请确认已安装 feishu-create-doc skill 并重启服务。" ) result = feishu_create_doc( markdown=report_markdown, title=title, folder_token=folder_token or "", wiki_space=wiki_space_id or "", ) if result.get("task_id"): # Async creation — poll doc_token = self._poll_doc_task(result["task_id"]) else: doc_token = result.get("document_id", "") url = f"https://aiplayer.feishu.cn/docx/{doc_token}" return { "document_id": doc_token, "url": url, } def _poll_doc_task(self, task_id: str, timeout: int = 30) -> str: """Poll async doc creation task until done.""" from feishu_create_doc import feishu_create_doc start = time.time() while time.time() - start < timeout: result = feishu_create_doc(task_id=task_id) status = result.get("status", "") if status == "success" or "document_id" in result: return result.get("document_id", result.get("doc_token", "")) time.sleep(2) raise ExportError("文档创建超时，请稍后重试。") # ─── Convenience functions ────────────────────────────────────────────────────── def export_excel(df: pd.DataFrame, path: str) -> str: exp = DataExporter(df) return exp.to_excel(path) def export_csv(df: pd.DataFrame, path: str) -> str: exp = DataExporter(df) return exp.to_csv(path) FILE:scripts/merger.py """ F5 · Multi-source data merge and join. Supports: - Exact join on key columns - Fuzzy join (fuzzywuzzy) when exact match fails - Auto-key detection Usage: from merger import DataMerger, MergeResult merger = DataMerger(sources) # sources = [(name, df), ...] result = merger.merge( how="left", on=[("手机号", "电话")], # [(left_col, right_col)] fuzzy_on=[("姓名", "客户名")], fuzzy_threshold=85, ) df_merged = result.df """ import re from typing import Dict, List, Optional, Tuple, Any from dataclasses import dataclass, field import pandas as pd # ─── Exceptions ───────────────────────────────────────────────────────────────── class MergeError(Exception): pass # ─── Results ─────────────────────────────────────────────────────────────────── @dataclass class MergeResult: df: pd.DataFrame left_name: str right_name: str matched_rows: int unmatched_left: int unmatched_right: int fuzzy_matched: int join_type: str def summary(self) -> str: return ( f"合并「{self.left_name}」+「{self.right_name}」\n" f"匹配行：{self.matched_rows} | 左表未匹配：{self.unmatched_left}" f" | 右表未匹配：{self.unmatched_right}\n" f"模糊匹配：{self.fuzzy_matched} 行" ) # ─── Merger ──────────────────────────────────────────────────────────────────── class DataMerger: """ Merge multiple DataFrames with exact and fuzzy join support. Parameters ---------- sources : List[Tuple[name, df]] """ def __init__(self, sources: List[Tuple[str, pd.DataFrame]]): if len(sources) < 2: raise MergeError("至少需要 2 个数据源才能合并。") self.sources = sources def merge( self, how: str = "inner", on: Optional[List[Tuple[str, str]]] = None, fuzzy_on: Optional[List[Tuple[str, str]]] = None, fuzzy_threshold: int = 85, suffix_left: str = "_x", suffix_right: str = "_y", ) -> MergeResult: """ Perform a join between the first two sources. Parameters ---------- how : "inner" | "left" | "right" | "outer" | "cross" on : list of (left_col, right_col) for exact join fuzzy_on : list of (left_col, right_col) for fuzzy join (applied after exact join fails) fuzzy_threshold: 0-100, fuzzy match score threshold suffix_left/right: suffix for overlapping columns """ if how not in ("inner", "left", "right", "outer", "cross"): raise MergeError(f"不支持的 join 类型：{how}。可选：inner/left/right/outer/cross") left_name, left_df = self.sources[0] right_name, right_df = self.sources[1] # Normalise: strip + lowercase column names left_df = left_df.copy() right_df = right_df.copy() left_df.columns = [str(c).strip() for c in left_df.columns] right_df.columns = [str(c).strip() for c in right_df.columns] # ── Auto-detect key columns if not provided ──────────────────────────── if not on and not fuzzy_on: on = self._auto_detect_keys(left_df, right_df) if not on and not fuzzy_on: raise MergeError( "未指定合并键，请通过 on= 参数指定要关联的列名，" "或使用 fuzzy_on= 进行模糊关联。" ) # ── Exact join ───────────────────────────────────────────────────────── df_exact, exact_matched = self._exact_merge( left_df, right_df, on or [], how, suffix_left, suffix_right ) # ── Fuzzy join on remaining rows ─────────────────────────────────────── fuzzy_matched = 0 if fuzzy_on: df_exact, fuzzy_matched = self._fuzzy_merge( df_exact, left_df, right_df, fuzzy_on, exact_matched, how, fuzzy_threshold, suffix_left, suffix_right ) # ── Compute stats ─────────────────────────────────────────────────────── matched = exact_matched + fuzzy_matched if how in ("left", "inner"): unmatched_left = len(left_df) - matched else: unmatched_left = 0 if how in ("right", "inner", "outer"): unmatched_right = len(right_df) - matched else: unmatched_right = 0 return MergeResult( df=df_exact, left_name=left_name, right_name=right_name, matched_rows=matched, unmatched_left=unmatched_left, unmatched_right=unmatched_right, fuzzy_matched=fuzzy_matched, join_type=how, ) # ─── Auto key detection ───────────────────────────────────────────────────── KEY_PATTERNS = { "phone": ["手机", "电话", "mobile", "phone", "tel"], "email": ["邮箱", "email", "mail"], "name": ["姓名", "name", "客户名", "用户名", "username"], "order": ["订单", "order", "order_no", "订单号"], "sku": ["sku", "商品编号", "产品编号"], "id": ["id", "编号", "用户id"], } def _auto_detect_keys( self, left: pd.DataFrame, right: pd.DataFrame, ) -> List[Tuple[str, str]]: """Find matching column pairs between left and right DataFrames.""" matches: List[Tuple[str, str]] = [] for pattern_name, keywords in self.KEY_PATTERNS.items(): for kw in keywords: kw_lower = kw.lower() left_cols = [c for c in left.columns if kw_lower in c.lower()] right_cols = [c for c in right.columns if kw_lower in c.lower()] if left_cols and right_cols: matches.append((left_cols[0], right_cols[0])) return matches # ─── Exact merge ──────────────────────────────────────────────────────────── def _exact_merge( self, left: pd.DataFrame, right: pd.DataFrame, on: List[Tuple[str, str]], how: str, sfx_l: str, sfx_r: str, ) -> Tuple[pd.DataFrame, int]: """Perform pandas merge on specified columns.""" if not on: return left, 0 left_keys = [pair[0] for pair in on] right_keys = [pair[1] for pair in on] # Rename right keys to match left for pandas merge right_renamed = right.rename(columns=dict(zip(right_keys, left_keys))) # Only keep right columns not in left (to avoid ambiguity) overlap = set(left.columns) & set(right_renamed.columns) - set(left_keys) right_dedup = right_renamed.drop(columns=list(overlap), errors="ignore") merged = left.merge( right_dedup, left_on=left_keys, right_on=left_keys, how=how, suffixes=(sfx_l, sfx_r), ) # Count matched rows (rows that got a match) matched = len(merged) return merged, matched # ─── Fuzzy merge ──────────────────────────────────────────────────────────── def _fuzzy_merge( self, merged_df: pd.DataFrame, left_orig: pd.DataFrame, right_orig: pd.DataFrame, fuzzy_on: List[Tuple[str, str]], already_matched: int, how: str, threshold: int, sfx_l: str, sfx_r: str, ) -> Tuple[pd.DataFrame, int]: """ For left rows with no match, try fuzzy match against right. Append fuzzy-matched rows to merged_df. """ try: from fuzzywuzzy import fuzz except ImportError: return merged_df, 0 if not fuzzy_on: return merged_df, 0 left_col, right_col = fuzzy_on[0] # Rows already matched have non-null right-side data # We identify unmatched by checking if right-side join columns are null right_cols_in_merged = [c for c in merged_df.columns if sfx_r in c] if not right_cols_in_merged: # No right columns present yet; treat all as unmatched unmatched_mask = pd.Series(True, index=merged_df.index) else: # If all right suffix cols are null → unmatched unmatched_mask = merged_df[right_cols_in_merged[0]].isna() unmatched_left = merged_df.loc[unmatched_mask, left_col].copy() right_vals = right_orig[right_col].astype(str).tolist() right_orig_index = right_orig.index.tolist() fuzzy_rows = [] fuzzy_count = 0 for idx, left_val in unmatched_left.items(): best_score = 0 best_row = None best_ridx = None for ri, rv in enumerate(right_vals): score = fuzz.ratio(str(left_val), str(rv)) if score > best_score and score >= threshold: best_score = score best_row = right_orig.iloc[ri] best_ridx = right_orig_index[ri] if best_row is not None: fuzzy_count += 1 row_data = merged_df.loc[idx].copy() # Add right columns that aren't already present for col in right_orig.columns: new_col = f"{col}{sfx_r}" if new_col not in row_data.index: row_data[new_col] = best_row[col] fuzzy_rows.append(row_data) if fuzzy_rows: fuzzy_df = pd.DataFrame(fuzzy_rows, index=[r.name for r in fuzzy_rows]) # Ensure same column order fuzzy_df = fuzzy_df.reindex(columns=merged_df.columns) merged_df = pd.concat([merged_df, fuzzy_df], ignore_index=True) return merged_df, fuzzy_count # ─── Convenience: merge all sources iteratively ────────────────────────────── def merge_all( self, on: Optional[List[Tuple[str, str]]] = None, fuzzy_on: Optional[List[Tuple[str, str]]] = None, fuzzy_threshold: int = 85, ) -> Tuple[pd.DataFrame, List[MergeResult]]: """ Merge all sources in order (left-fold). Returns the final DataFrame and list of per-step results. """ if len(self.sources) == 2: result = self.merge( how="outer", on=on, fuzzy_on=fuzzy_on, fuzzy_threshold=fuzzy_threshold, ) return result.df, [result] # First merge temp_sources = self.sources[:2] merger = DataMerger(temp_sources) first = merger.merge(how="outer", on=on, fuzzy_on=fuzzy_on, fuzzy_threshold=fuzzy_threshold) results = [first] current = ("merged_0", first.df) for i in range(2, len(self.sources)): _, next_src = self.sources[i] temp_sources = [current, (self.sources[i][0], next_src)] merger = DataMerger(temp_sources) step = merger.merge(how="outer", on=on, fuzzy_on=fuzzy_on, fuzzy_threshold=fuzzy_threshold) results.append(step) current = (f"merged_{i}", step.df) return current[1], results

Data Cleaner AI

Comments (0)