Python连载05|综合实战:爬虫、Excel自动化与数据分析

import requestsimport jsonimport osimport timeclass DoubanSpider:    """豆瓣电影Top250爬虫"""    def __init__(self):        self.base_url = "https://movie.douban.com/top250"        self.headers = {            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"        }        self.movies = []    def fetch_page(self, start):        """抓取一页数据（每页25条）"""        params = {"start": start, "filter": ""}        try:            resp = requests.get(                self.base_url, params=params,                headers=self.headers, timeout=10            )            return resp.text        except requests.RequestException as e:            print(f"❌ 请求失败：{e}")            return None    def parse_html(self, html):        """从HTML中提取电影信息（简化版演示）"""        # 实际项目中用 BeautifulSoup 或 re 正则提取        # 这里演示数据结构        movie = {            "title": "肖申克的救赎",            "rating": 9.7,            "year": 1994,            "director": "弗兰克·德拉邦特"        }        self.movies.append(movie)    def run(self, pages=2):        """运行爬虫"""        print("🕷 开始爬取豆瓣Top250...")        for page in range(pages):            start = page * 25            print(f"  正在爬取第 {page+1} 页...")            html = self.fetch_page(start)            if html:                self.parse_html(html)            time.sleep(2)  # 礼貌爬取，间隔2秒        self.save_results()        print(f"✅ 爬取完成，共获取 {len(self.movies)} 部电影")    def save_results(self):        """保存结果到JSON和CSV"""        os.makedirs("output", exist_ok=True)        # 保存JSON        with open("output/movies.json", "w", encoding="utf-8") as f:            json.dump(self.movies, f, ensure_ascii=False, indent=2)        # 保存CSV        import csv        with open("output/movies.csv", "w", encoding="utf-8-sig", newline="") as f:            writer = csv.DictWriter(f, fieldnames=self.movies[0].keys())            writer.writeheader()            writer.writerows(self.movies)# 运行爬虫spider = DoubanSpider()spider.run(pages=2)

import pandas as pdimport osclass SalarySplitter:    """工资条自动拆分工具"""    def __init__(self, source_file):        self.source_file = source_file        self.output_dir = "output/salary_slips"        os.makedirs(self.output_dir, exist_ok=True)    def load_data(self):        """读取工资总表"""        print(f"📥 读取文件：{self.source_file}")        self.df = pd.read_excel(self.source_file)        print(f"  共 {len(self.df)} 条记录\n")    def split_and_save(self):        """拆分并保存每个人的工资条"""        for index, row in self.df.iterrows():            name = row["姓名"]            # 每个人一个DataFrame            slip = pd.DataFrame([row])            # 保存为单独的Excel文件            filename = f"{name}_工资条.xlsx"            filepath = os.path.join(self.output_dir, filename)            slip.to_excel(filepath, index=False, sheet_name="工资条")            print(f"  ✅ {name} → {filename}")    def generate_summary(self):        """生成汇总统计"""        print(f"\n📊 工资汇总：")        print(f"  总人数：{len(self.df)}")        print(f"  平均工资：{self.df['实发工资'].mean():.2f} 元")        print(f"  最高工资：{self.df['实发工资'].max():.2f} 元")        print(f"  最低工资：{self.df['实发工资'].min():.2f} 元")        # 工资排名        ranked = self.df.sort_values("实发工资", ascending=False)        print(f"\n  工资前3名：")        for i, (_, row) in enumerate(ranked.head(3).iterrows()):            print(f"    {i+1}. {row['姓名']}：{row['实发工资']:.2f}元")    def run(self):        """一键执行"""        self.load_data()        self.split_and_save()        self.generate_summary()        print(f"\n📁 所有文件已保存到：{self.output_dir}")# ===== 模拟工资数据 =====salary_data = pd.DataFrame({    "姓名": ["张三", "李四", "王五", "赵六", "钱七"],    "部门": ["技术部", "市场部", "技术部", "财务部", "市场部"],    "基本工资": [15000, 12000, 18000, 13000, 11000],    "绩效奖金": [3000, 2000, 5000, 1500, 2500],    "社保扣除": [1200, 960, 1440, 1040, 880],})salary_data["实发工资"] = salary_data["基本工资"] + salary_data["绩效奖金"] - salary_data["社保扣除"]salary_data.to_excel("工资总表.xlsx", index=False)# 运行拆分工具splitter = SalarySplitter("工资总表.xlsx")splitter.run()

import pandas as pdimport osimport time# 装饰器：自动计时def timer(func):    def wrapper(*args, **kwargs):        start = time.time()        result = func(*args, **kwargs)        print(f"  ⏱ {func.__name__} 完成，耗时 {time.time()-start:.3f}s")        return result    return wrapperclass SalesAnalyzer:    """销售数据分析器"""    def __init__(self, output_dir="output/sales_report"):        self.output_dir = output_dir        os.makedirs(output_dir, exist_ok=True)    @timer    def load_data(self):        """加载销售数据"""        # 模拟全年销售数据        import random        products = ["手机", "耳机", "电脑", "平板", "手表"]        regions = ["华北", "华东", "华南", "西部"]        months = range(1, 13)        records = []        for month in months:            for _ in range(20):  # 每月20条记录                records.append({                    "月份": month,                    "产品": random.choice(products),                    "区域": random.choice(regions),                    "销量": random.randint(10, 200),                    "金额": None                })        records[-1]["金额"] = None  # 故意造一条缺失值        self.df = pd.DataFrame(records)        # 计算金额 = 销量 * 单价（简化）        price_map = {"手机": 5000, "耳机": 500, "电脑": 8000, "平板": 3000, "手表": 2000}        self.df["金额"] = self.df["销量"] * self.df["产品"].map(price_map)        print(f"  📊 加载 {len(self.df)} 条销售记录")    @timer    def clean_data(self):        """数据清洗"""        before = len(self.df)        self.df = self.df.dropna()  # 删除缺失值        self.df = self.df.drop_duplicates()  # 删除重复值        print(f"  🧹 清洗完成：{before} → {len(self.df)} 条")    @timer    def analyze(self):        """多维度分析"""        print("\n📋 === 销售数据分析报告 ===\n")        # 1. 月度销售趋势        monthly = self.df.groupby("月份")["金额"].sum()        print("📈 月度销售金额：")        for month, amount in monthly.items():            bar = "█" * int(amount / 100000)            print(f"  {month:>2d}月：{amount:>10,.0f}元 {bar}")        # 2. 产品销量排行        product_sales = self.df.groupby("产品")["销量"].sum().sort_values(ascending=False)        print(f"\n🏆 产品销量排行：")        for i, (product, sales) in enumerate(product_sales.items()):            print(f"  {i+1}. {product}：{sales}件")        # 3. 区域销售占比        region_sales = self.df.groupby("区域")["金额"].sum()        total = region_sales.sum()        print(f"\n🗺 区域销售占比：")        for region, amount in region_sales.sort_values(ascending=False).items():            pct = amount / total * 100            print(f"  {region}：{amount:>10,.0f}元 ({pct:.1f}%)")        # 4. 保存分析结果        monthly.to_csv(os.path.join(self.output_dir, "月度趋势.csv"))        product_sales.to_csv(os.path.join(self.output_dir, "产品排行.csv"))        region_sales.to_csv(os.path.join(self.output_dir, "区域占比.csv"))        print(f"\n💾 分析结果已保存到 {self.output_dir}/")    @timer    def generate_chart(self):        """生成图表（需要 matplotlib）"""        # pip install matplotlib        import matplotlib        matplotlib.use("Agg")  # 无界面模式        import matplotlib.pyplot as plt        # 月度趋势折线图        monthly = self.df.groupby("月份")["金额"].sum()        fig, ax = plt.subplots(figsize=(10, 5))        ax.plot(monthly.index, monthly.values, "o-", color="#306090", linewidth=2)        ax.set_title("月度销售趋势", fontsize=14)        ax.set_xlabel("月份")        ax.set_ylabel("销售金额（元）")        ax.grid(True, alpha=0.3)        fig.savefig(os.path.join(self.output_dir, "月度趋势.png"), dpi=150)        print(f"  📊 图表已保存：月度趋势.png")    def run(self):        """一键执行全流程"""        print("🚀 启动销售数据分析...\n")        self.load_data()        self.clean_data()        self.analyze()        self.generate_chart()        print("\n✅ 分析完成！")# 运行分析器analyzer = SalesAnalyzer()analyzer.run()