Upload data1/reporting/stage_a_stats.py with huggingface_hub

data1/reporting/stage_a_stats.py

@@ -0,0 +1,368 @@
+"""
+Stage A: 流式统计 repos_searched.csv (130万+行)
+统计缺失率、by_keyword/by_language/by_license/by_year、stars/forks/open_issues/size分布
+"""
+import csv
+import sys
+from collections import defaultdict, Counter
+from datetime import datetime
+import json
+from pathlib import Path
+from tqdm import tqdm
+import statistics
+import re
+
+csv.field_size_limit(sys.maxsize)
+
+
+class StageAStats:
+    def __init__(self, csv_path, output_dir):
+        self.csv_path = csv_path
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        
+        # 统计数据容器
+        self.stats = {
+            'total_records': 0,
+            'unique_repos': set(),
+            'unique_owners': set(),
+            'unique_keywords': set(),
+            'by_keyword': defaultdict(lambda: {'count': 0, 'repos': set()}),
+            'by_language': defaultdict(int),
+            'by_license': defaultdict(int),
+            'by_year': defaultdict(int),
+            'by_year_month': defaultdict(int),
+            'missing': defaultdict(int),
+            'topics_stats': {'counts': [], 'missing': 0},
+            'desc_lengths': [],
+            'stars': [],
+            'forks': [],
+            'open_issues': [],
+            'size': [],
+            'archived': {'yes': 0, 'no': 0},
+            'has_wiki': {'yes': 0, 'no': 0},
+            'default_branch': defaultdict(int),
+            'repo_keyword_count': defaultdict(int),  # 每个repo命中多少个keyword
+            'topics_list': [],  # 所有topics
+        }
+        
+    def parse_date(self, date_str):
+        """解析日期字符串"""
+        if not date_str or date_str.strip() == '':
+            return None
+        try:
+            # ISO format: 2017-04-26T11:03:50Z
+            date_str = date_str.replace('Z', '+00:00')
+            return datetime.fromisoformat(date_str.replace('Z', '+00:00'))
+        except:
+            return None
+    
+    def safe_int(self, val):
+        """安全转换为整数"""
+        if not val or val == '':
+            return None
+        try:
+            return int(float(val))
+        except:
+            return None
+    
+    def safe_bool(self, val):
+        """安全转换为布尔值"""
+        if not val or val == '':
+            return None
+        val = str(val).strip().lower()
+        return val in ('true', 'yes', '1', 't')
+    
+    def is_empty(self, val):
+        """判断字段是否为空"""
+        if val is None:
+            return True
+        val = str(val).strip()
+        return val == '' or val.lower() == 'none'
+    
+    def process_row(self, row):
+        """处理单行数据"""
+        self.stats['total_records'] += 1
+        
+        # 基础字段
+        keyword = row.get('keyword', '').strip()
+        full_name = row.get('full_name', '').strip()
+        owner = row.get('owner', '').strip()
+        language = row.get('language', '').strip()
+        license_val = row.get('license', '').strip()
+        topics = row.get('topics', '').strip()
+        description = row.get('description', '').strip()
+        default_branch = row.get('default_branch', '').strip()
+        
+        # 唯一值统计
+        if keyword:
+            self.stats['unique_keywords'].add(keyword)
+        if full_name:
+            self.stats['unique_repos'].add(full_name)
+            self.stats['repo_keyword_count'][full_name] += 1
+        if owner:
+            self.stats['unique_owners'].add(owner)
+        
+        # by_keyword
+        if keyword:
+            self.stats['by_keyword'][keyword]['count'] += 1
+            if full_name:
+                self.stats['by_keyword'][keyword]['repos'].add(full_name)
+        
+        # 缺失率统计
+        for field in ['description', 'language', 'topics', 'license']:
+            if self.is_empty(row.get(field, '')):
+                self.stats['missing'][field] += 1
+        
+        # language
+        if language:
+            self.stats['by_language'][language] += 1
+        else:
+            self.stats['by_language']['<empty>'] += 1
+        
+        # license
+        if license_val:
+            self.stats['by_license'][license_val] += 1
+        else:
+            self.stats['by_license']['<empty>'] += 1
+        
+        # default_branch
+        if default_branch:
+            self.stats['default_branch'][default_branch] += 1
+        
+        # topics统计
+        if self.is_empty(topics):
+            self.stats['topics_stats']['missing'] += 1
+        else:
+            topics_list = [t.strip() for t in topics.split(',') if t.strip()]
+            self.stats['topics_stats']['counts'].append(len(topics_list))
+            self.stats['topics_list'].extend(topics_list)
+        
+        # description长度
+        if description:
+            self.stats['desc_lengths'].append(len(description))
+        
+        # 数值字段
+        stars = self.safe_int(row.get('stars'))
+        if stars is not None:
+            self.stats['stars'].append(stars)
+            
+        forks = self.safe_int(row.get('forks'))
+        if forks is not None:
+            self.stats['forks'].append(forks)
+            
+        open_issues = self.safe_int(row.get('open_issues'))
+        if open_issues is not None:
+            self.stats['open_issues'].append(open_issues)
+            
+        size = self.safe_int(row.get('size'))
+        if size is not None:
+            self.stats['size'].append(size)
+        
+        # 布尔字段
+        archived = self.safe_bool(row.get('archived'))
+        if archived is not None:
+            if archived:
+                self.stats['archived']['yes'] += 1
+            else:
+                self.stats['archived']['no'] += 1
+        
+        has_wiki = self.safe_bool(row.get('has_wiki'))
+        if has_wiki is not None:
+            if has_wiki:
+                self.stats['has_wiki']['yes'] += 1
+            else:
+                self.stats['has_wiki']['no'] += 1
+        
+        # 时间字段
+        created_at = self.parse_date(row.get('created_at'))
+        if created_at:
+            year = created_at.year
+            self.stats['by_year'][year] += 1
+            year_month = f"{year}-{created_at.month:02d}"
+            self.stats['by_year_month'][year_month] += 1
+    
+    def process_csv(self):
+        """流式处理CSV文件"""
+        print(f"Processing {self.csv_path}...")
+        
+        with open(self.csv_path, 'r', encoding='utf-8', errors='replace') as f:
+            reader = csv.DictReader(f)
+            for row in tqdm(reader, desc="Processing repos_searched.csv"):
+                self.process_row(row)
+        
+        # 转换set为计数
+        self.stats['unique_repos_count'] = len(self.stats['unique_repos'])
+        self.stats['unique_owners_count'] = len(self.stats['unique_owners'])
+        self.stats['unique_keywords_count'] = len(self.stats['unique_keywords'])
+        
+        # 转换by_keyword的repos set为计数
+        for kw in self.stats['by_keyword']:
+            self.stats['by_keyword'][kw]['unique_repos'] = len(self.stats['by_keyword'][kw]['repos'])
+            del self.stats['by_keyword'][kw]['repos']  # 释放内存
+    
+    def save_summary(self):
+        """保存总体汇总"""
+        summary = {
+            'total_records': self.stats['total_records'],
+            'unique_repos': self.stats['unique_repos_count'],
+            'unique_owners': self.stats['unique_owners_count'],
+            'unique_keywords': self.stats['unique_keywords_count'],
+            'missing_rates': {
+                field: self.stats['missing'][field] / self.stats['total_records'] * 100
+                for field in ['description', 'language', 'topics', 'license']
+            },
+            'archived_ratio': self.stats['archived']['yes'] / (self.stats['archived']['yes'] + self.stats['archived']['no']) * 100 if (self.stats['archived']['yes'] + self.stats['archived']['no']) > 0 else 0,
+            'has_wiki_ratio': self.stats['has_wiki']['yes'] / (self.stats['has_wiki']['yes'] + self.stats['has_wiki']['no']) * 100 if (self.stats['has_wiki']['yes'] + self.stats['has_wiki']['no']) > 0 else 0,
+        }
+        
+        # 添加分位数统计
+        if self.stats['stars']:
+            sorted_stars = sorted(self.stats['stars'])
+            summary['stars'] = {
+                'min': sorted_stars[0],
+                'p50': sorted_stars[len(sorted_stars)//2],
+                'p90': sorted_stars[int(len(sorted_stars)*0.9)],
+                'p99': sorted_stars[int(len(sorted_stars)*0.99)],
+                'max': sorted_stars[-1],
+                'mean': statistics.mean(sorted_stars),
+            }
+        
+        if self.stats['topics_stats']['counts']:
+            counts = self.stats['topics_stats']['counts']
+            summary['topics_per_repo'] = {
+                'mean': statistics.mean(counts),
+                'median': statistics.median(counts),
+                'max': max(counts),
+            }
+        
+        if self.stats['desc_lengths']:
+            lengths = self.stats['desc_lengths']
+            summary['description_length'] = {
+                'mean': statistics.mean(lengths),
+                'median': statistics.median(lengths),
+                'max': max(lengths),
+            }
+        
+        with open(self.output_dir / 'summary_overall.json', 'w', encoding='utf-8') as f:
+            json.dump(summary, f, indent=2, ensure_ascii=False)
+        
+        # 保存CSV格式
+        import pandas as pd
+        summary_df = pd.DataFrame([{
+            'metric': k,
+            'value': v if not isinstance(v, dict) else json.dumps(v)
+        } for k, v in summary.items()])
+        summary_df.to_csv(self.output_dir / 'summary_overall.csv', index=False)
+    
+    def save_by_keyword(self):
+        """保存按keyword的统计"""
+        rows = []
+        for kw, data in self.stats['by_keyword'].items():
+            rows.append({
+                'keyword': kw,
+                'count': data['count'],
+                'unique_repos': data['unique_repos'],
+            })
+        
+        import pandas as pd
+        df = pd.DataFrame(rows)
+        df = df.sort_values('count', ascending=False)
+        df.to_csv(self.output_dir / 'by_keyword.csv', index=False)
+    
+    def save_by_language(self):
+        """保存按语言的统计"""
+        import pandas as pd
+        df = pd.DataFrame([
+            {'language': lang, 'count': count}
+            for lang, count in sorted(self.stats['by_language'].items(), key=lambda x: -x[1])
+        ])
+        df.to_csv(self.output_dir / 'by_language.csv', index=False)
+    
+    def save_by_license(self):
+        """保存按license的统计"""
+        import pandas as pd
+        df = pd.DataFrame([
+            {'license': lic, 'count': count}
+            for lic, count in sorted(self.stats['by_license'].items(), key=lambda x: -x[1])
+        ])
+        df.to_csv(self.output_dir / 'by_license.csv', index=False)
+    
+    def save_by_year(self):
+        """保存按年份的统计"""
+        import pandas as pd
+        df = pd.DataFrame([
+            {'year': year, 'count': count}
+            for year, count in sorted(self.stats['by_year'].items())
+        ])
+        df.to_csv(self.output_dir / 'by_year.csv', index=False)
+        
+        # 按年月
+        df_month = pd.DataFrame([
+            {'year_month': ym, 'count': count}
+            for ym, count in sorted(self.stats['by_year_month'].items())
+        ])
+        df_month.to_csv(self.output_dir / 'by_year_month.csv', index=False)
+    
+    def save_top_repos(self):
+        """保存Top仓库（需要重新读取，这里先保存stars列表）"""
+        # 由于是流式处理，Top仓库需要单独处理或二次扫描
+        # 这里先保存stars分布数据
+        if self.stats['stars']:
+            import pandas as pd
+            df = pd.DataFrame({
+                'stars': sorted(self.stats['stars'], reverse=True)
+            })
+            df.to_csv(self.output_dir / 'stars_distribution.csv', index=False)
+    
+    def save_top_topics(self):
+        """保存Top topics"""
+        topic_counter = Counter(self.stats['topics_list'])
+        import pandas as pd
+        df = pd.DataFrame([
+            {'topic': topic, 'count': count}
+            for topic, count in topic_counter.most_common(100)
+        ])
+        df.to_csv(self.output_dir / 'top_topics.csv', index=False)
+    
+    def save_repo_keyword_counts(self):
+        """保存每个仓库命中的keyword数量分布"""
+        import pandas as pd
+        counts = list(self.stats['repo_keyword_count'].values())
+        df = pd.DataFrame({
+            'keyword_count': counts
+        })
+        df.to_csv(self.output_dir / 'repo_keyword_count_distribution.csv', index=False)
+    
+    def save_default_branch(self):
+        """保存default_branch分布"""
+        import pandas as pd
+        df = pd.DataFrame([
+            {'branch': branch, 'count': count}
+            for branch, count in sorted(self.stats['default_branch'].items(), key=lambda x: -x[1])
+        ])
+        df.to_csv(self.output_dir / 'default_branch_distribution.csv', index=False)
+    
+    def run(self):
+        """执行完整流程"""
+        print("Stage A: Processing repos_searched.csv...")
+        self.process_csv()
+        print("Saving results...")
+        self.save_summary()
+        self.save_by_keyword()
+        self.save_by_language()
+        self.save_by_license()
+        self.save_by_year()
+        self.save_top_repos()
+        self.save_top_topics()
+        self.save_repo_keyword_counts()
+        self.save_default_branch()
+        print(f"Stage A complete! Results saved to {self.output_dir}")
+
+
+if __name__ == "__main__":
+    csv_path = "/home/weifengsun/tangou1/domain_code/src/workdir/repos_searched.csv"
+    output_dir = "/home/weifengsun/tangou1/domain_code/src/workdir/reporting/stage_a"
+    stats = StageAStats(csv_path, output_dir)
+    stats.run()
+