每天重复3小时的活，Python帮我10分钟搞定

importpandasaspd
importos
fromopenpyxlimportload_workbook
fromopenpyxl.stylesimportFont, Alignment, Border, Side

defauto_excel_report():
    # 1. 自动读取所有部门的Excel
    department_files = [fforfinos.listdir('departments/') iff.endswith('.xlsx')]
    
    all_data = []
    forfileindepartment_files:
        # 读取每个部门的数据
        df = pd.read_excel(f'departments/{file}', sheet_name='月度数据')
        # 自动提取需要的列
        extracted = df[['部门', '销售额', '成本', '利润']].copy()
        extracted['月份'] = pd.Timestamp.now().strftime('%Y-%m')
        all_data.append(extracted)
    
    # 2. 合并数据
    total_df = pd.concat(all_data, ignore_index=True)
    
    # 3. 计算汇总
    summary = {
        '总销售额': total_df['销售额'].sum(),
        '总成本': total_df['成本'].sum(),
        '总利润': total_df['利润'].sum(),
        '平均利润率': (total_df['利润'].sum() /total_df['销售额'].sum() *100)
    }
    
    # 4. 保存到总表（带格式）
    withpd.ExcelWriter('月度汇总报告.xlsx', engine='openpyxl') aswriter:
        total_df.to_excel(writer, sheet_name='明细数据', index=False)
        
        # 获取工作表对象用于格式设置
        workbook = writer.book
        worksheet = writer.sheets['明细数据']
        
        # 设置标题格式
        header_font = Font(bold=True, color="FFFFFF")
        header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
        
        forcellinworksheet[1]:
            cell.font = header_font
            cell.fill = header_font
            cell.alignment = Alignment(horizontal='center')
        
        # 自动调整列宽
        forcolumninworksheet.columns:
            max_length = 0
            column_letter = column[0].column_letter
            forcellincolumn:
                try:
                    iflen(str(cell.value)) >max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length+2, 50)
            worksheet.column_dimensions[column_letter].width = adjusted_width
    
    print(f"报表生成完成！耗时: 5分钟")
    print(f"汇总数据: {summary}")
    
    # 5. 自动发送邮件（可选）
    # send_email_with_attachment('月度汇总报告.xlsx')

importos
importshutil
fromdatetimeimportdatetime
frompathlibimportPath

defauto_file_organizer(source_folder='下载/'):
    # 定义分类规则
    categories = {
        '文档': ['.pdf', '.doc', '.docx', '.txt', '.md'],
        '表格': ['.xlsx', '.xls', '.csv'],
        '图片': ['.jpg', '.jpeg', '.png', '.gif', '.bmp'],
        '代码': ['.py', '.js', '.java', '.cpp', '.html', '.css'],
        '压缩包': ['.zip', '.rar', '.7z', '.tar.gz']
    }
    
    # 创建分类文件夹
    forcategoryincategories.keys():
        os.makedirs(f'整理后/{category}', exist_ok=True)
    
    # 遍历并分类文件
    forfilenameinos.listdir(source_folder):
        filepath = os.path.join(source_folder, filename)
        
        ifos.path.isfile(filepath):
            # 获取文件扩展名
            ext = os.path.splitext(filename)[1].lower()
            
            # 查找对应的分类
            target_category = '其他'  # 默认分类
            forcategory, extensionsincategories.items():
                ifextinextensions:
                    target_category = category
                    break
            
            # 生成新文件名（日期+原名）
            today = datetime.now().strftime('%Y%m%d')
            new_filename = f"{today}_{filename}"
            
            # 移动文件
            target_path = f'整理后/{target_category}/{new_filename}'
            shutil.move(filepath, target_path)
            
            print(f"已移动: {filename} → {target_category}/{new_filename}")
    
    # 生成整理报告
    report = {}
    forcategoryincategories.keys():
        category_path = f'整理后/{category}'
        ifos.path.exists(category_path):
            count = len([fforfinos.listdir(category_path) 
                        ifos.path.isfile(os.path.join(category_path, f))])
            report[category] = count
    
    print(f"\n整理完成！")
    forcategory, countinreport.items():
        print(f"{category}: {count}个文件")

# 设置定时任务，每天下午5点自动整理
# 在Linux/Mac: crontab -e 添加: 0 17 * * * python /path/to/organizer.py
# 在Windows: 使用任务计划程序

importhashlib

defremove_duplicates(folder_path):
    """删除重复文件"""
    hashes = {}
    
    forroot, dirs, filesinos.walk(folder_path):
        forfilenameinfiles:
            filepath = os.path.join(root, filename)
            
            # 计算文件哈希值
            withopen(filepath, 'rb') asf:
                file_hash = hashlib.md5(f.read()).hexdigest()
            
            # 如果哈希值已存在，删除重复文件
            iffile_hashinhashes:
                print(f"删除重复文件: {filepath}")
                os.remove(filepath)
            else:
                hashes[file_hash] = filepath

importrequests
frombs4importBeautifulSoup
importpandasaspd
importschedule
importtime
fromdatetimeimportdatetime

deffetch_product_prices():
    """抓取多个电商网站的价格"""
    
    # 定义要监控的产品和网站
    products = {
        'iPhone15': {
            '京东': 'https://item.jd.com/10012345678.html',
            '天猫': 'https://detail.tmall.com/item.htm?id=123456789',
            '苏宁': 'https://product.suning.com/123456789.html'
        },
        '小米电视': {
            '京东': 'https://item.jd.com/10023456789.html',
            '天猫': 'https://detail.tmall.com/item.htm?id=234567890'
        }
    }
    
    all_prices = []
    
    forproduct_name, websitesinproducts.items():
        forsite_name, urlinwebsites.items():
            try:
                # 发送请求
                headers = {
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                }
                response = requests.get(url, headers=headers, timeout=10)
                
                # 解析页面（每个网站结构不同，需要单独处理）
                soup = BeautifulSoup(response.text, 'html.parser')
                
                # 京东价格提取
                if'jd.com'inurl:
                    price_element = soup.find('span', class_='price J-p-10012345678')
                    price = float(price_element.text.strip().replace('¥', '')) ifprice_elementelseNone
                
                # 天猫价格提取
                elif'tmall.com'inurl:
                    price_element = soup.find('span', class_='tm-price')
                    price = float(price_element.text.strip()) ifprice_elementelseNone
                
                # 记录价格
                ifprice:
                    all_prices.append({
                        '时间': datetime.now().strftime('%Y-%m-%d %H:%M'),
                        '产品': product_name,
                        '平台': site_name,
                        '价格': price,
                        '网址': url
                    })
                    print(f"{product_name}在{site_name}的价格: ¥{price}")
                
            exceptExceptionase:
                print(f"抓取{site_name}的{product_name}价格失败: {e}")
    
    # 保存到Excel
    ifall_prices:
        df = pd.DataFrame(all_prices)
        
        # 如果文件已存在，追加数据
        ifos.path.exists('价格监控.xlsx'):
            existing_df = pd.read_excel('价格监控.xlsx')
            updated_df = pd.concat([existing_df, df], ignore_index=True)
        else:
            updated_df = df
        
        # 保存数据
        updated_df.to_excel('价格监控.xlsx', index=False)
        
        # 生成价格分析报告
        generate_price_report(updated_df)
    
    returnall_prices

defgenerate_price_report(df):
    """生成价格分析报告"""
    
    # 按产品和平台分组
    latest_prices = df.sort_values('时间').groupby(['产品', '平台']).last().reset_index()
    
    # 计算每个产品的最低价格和平台
    product_summary = []
    forproductindf['产品'].unique():
        product_data = latest_prices[latest_prices['产品'] == product]
        min_price = product_data['价格'].min()
        min_platform = product_data[product_data['价格'] == min_price]['平台'].iloc[0]
        
        product_summary.append({
            '产品': product,
            '最低价': min_price,
            '最低价平台': min_platform,
            '监控平台数': len(product_data),
            '建议购买平台': min_platform
        })
    
    # 保存报告
    summary_df = pd.DataFrame(product_summary)
    summary_df.to_excel('价格分析报告.xlsx', index=False)
    
    print("\n=== 价格分析报告 ===")
    foriteminproduct_summary:
        print(f"{item['产品']}: 最低¥{item['最低价']} ({item['最低价平台']})")

defrun_daily_monitor():
    """每天定时运行监控"""
    print(f"{datetime.now().strftime('%Y-%m-%d %H:%M')} 开始价格监控...")
    fetch_product_prices()
    print("价格监控完成！")

# 设置定时任务：每天上午10点和下午4点各运行一次
schedule.every().day.at("10:00").do(run_daily_monitor)
schedule.every().day.at("16:00").do(run_daily_monitor)

print("价格监控系统已启动，按Ctrl+C停止")
whileTrue:
    schedule.run_pending()
    time.sleep(60)

importimaplib
importemail
fromemail.headerimportdecode_header
importre
fromdatetimeimportdatetime, timedelta

classAutoEmailProcessor:
    def__init__(self, email_address, password):
        self.email_address = email_address
        self.password = password
        
    defconnect_to_mailbox(self):
        """连接到邮箱"""
        self.mail = imaplib.IMAP4_SSL("imap.example.com")
        self.mail.login(self.email_address, self.password)
        self.mail.select("inbox")
    
    deffetch_unread_emails(self):
        """获取未读邮件"""
        status, messages = self.mail.search(None, 'UNSEEN')
        email_ids = messages[0].split()
        
        emails = []
        foremail_idinemail_ids:
            status, msg_data = self.mail.fetch(email_id, '(RFC822)')
            email_body = msg_data[0][1]
            msg = email.message_from_bytes(email_body)
            
            # 解析邮件信息
            subject = self._decode_header(msg["Subject"])
            from_ = self._decode_header(msg["From"])
            date = msg["Date"]
            body = self._get_email_body(msg)
            
            emails.append({
                'id': email_id.decode(),
                'subject': subject,
                'from': from_,
                'date': date,
                'body': body
            })
        
        returnemails
    
    defauto_classify_and_reply(self):
        """自动分类和回复邮件"""
        emails = self.fetch_unread_emails()
        
        # 定义分类规则和回复模板
        rules = [
            {
                'keywords': ['报价', '价格', '多少钱'],
                'category': '价格咨询',
                'reply_template': '''尊敬的客户：

感谢您对我们产品的关注！

关于{product}的价格信息，请参考我们的官网报价页面：https://example.com/pricing

如果您需要批量采购或有特殊需求，请联系我们的销售团队。

祝好！
{company}团队'''
            },
            {
                'keywords': ['技术支持', '故障', '问题', '帮助'],
                'category': '技术支持',
                'reply_template': '''尊敬的客户：

感谢您联系我们！

我们已经收到您的技术支持请求，工单编号：{ticket_id}

我们的技术支持团队将在24小时内联系您。

您也可以通过以下方式获取帮助：
1. 知识库：https://help.example.com
2. 在线客服：工作日9:00-18:00

{ticket_details}'''
            }
        ]
        
        foremail_infoinemails:
            category = '其他'
            reply_template = None
            
            # 根据邮件内容分类
            email_text = f"{email_info['subject']} {email_info['body']}".lower()
            
            forruleinrules:
                ifany(keywordinemail_textforkeywordinrule['keywords']):
                    category = rule['category']
                    reply_template = rule['reply_template']
                    break
            
            # 生成回复
            ifreply_template:
                reply_content = self._generate_reply(email_info, reply_template, category)
                self._send_reply(email_info['from'], 
                               f"Re: {email_info['subject']}", 
                               reply_content)
                
                print(f"已自动回复邮件: {email_info['subject']} ({category})")
            
            # 标记为已读
            self.mail.store(email_info['id'], '+FLAGS', '\\Seen')
    
    def_generate_reply(self, email_info, template, category):
        """生成回复内容"""
        # 从邮件中提取产品名称
        product_match = re.search(r'产品[：:]?\s*([^\s，,。]+)', email_info['body'])
        product = product_match.group(1) ifproduct_matchelse"相关产品"
        
        # 生成工单ID
        ticket_id = f"TICKET-{datetime.now().strftime('%Y%m%d')}-{hash(email_info['id'])%1000:03d}"
        
        # 填充模板
        reply = template.format(
            product=product,
            ticket_id=ticket_id,
            company="XX科技",
            ticket_details=f"问题描述：{email_info['body'][:100]}..."
        )
        
        returnreply
    
    def_send_reply(self, to_address, subject, content):
        """发送回复邮件"""
        # 这里需要实现SMTP发送逻辑
        pass
    
    def_decode_header(self, header):
        """解码邮件头"""
        ifheader:
            decoded = decode_header(header)
            return''.join([str(t[0], t[1] or'utf-8') ifisinstance(t[0], bytes) elset[0] 
                           fortindecoded])
        return""
    
    def_get_email_body(self, msg):
        """获取邮件正文"""
        ifmsg.is_multipart():
            forpartinmsg.walk():
                content_type = part.get_content_type()
                content_disposition = str(part.get("Content-Disposition"))
                
                ifcontent_type == "text/plain"and"attachment"notincontent_disposition:
                    body = part.get_payload(decode=True).decode()
                    returnbody
        else:
            content_type = msg.get_content_type()
            ifcontent_type == "text/plain":
                body = msg.get_payload(decode=True).decode()
                returnbody
        
        return""

# 核心库推荐
-文件操作: os, shutil, pathlib
-Excel处理: pandas, openpyxl
-网页抓取: requests, BeautifulSoup
-邮件处理: imaplib, smtplib
-定时任务: schedule, apscheduler