Excel转通达信自定义数据2.0优化版(pandas+tkinter+含业绩报)

self.df_data = pd.read_html(file_path, header=0)[0]

pip install lxml

with open('全部Ａ股_20260522_2.xls', 'rb') as f:    raw = f.read()    text = raw.decode('gb18030', errors='replace')    print(text)

# -*- coding: utf-8 -*-# @Author : 小红牛# 微信公众号：gxzfp888import tkinter as tkfrom tkinter import ttk, filedialog, messagebox, scrolledtextfrom datetime import datetime, timedelta, date as date_class   # 明确导入 date 类import pandas as pdimport requestsimport threadingimport os# ==================== 第一部分：Excel转通达信 ====================class StockDataProcessor:    """Excel转通达信自定义数据工具"""    def __init__(self, parent_frame):        self.parent = parent_frame        self.df_data = None        self.current_headers = []        self.excel_path = ""        self.data_type = tk.StringVar(value="external")        # 创建 Ex数据 文件夹（如果不存在）        self.ex_dir = os.path.join(os.getcwd(), "Ex数据")        os.makedirs(self.ex_dir, exist_ok=True)        self.create_widgets()    # ---------- 工具函数 ----------    @staticmethod    def extract_code(s):        if s[0].isalpha():            return s[-6:]        elif s[-1].isalpha():            return s[:6]        else:            return s    @staticmethod    def get_stock_type(code):        if code.startswith(('60', '68')):            return '1'        elif code.startswith(('30', '00')):            return '0'        elif code.startswith('92'):            return '2'        else:            return '3'    def fm_stock_code(self, raw_code):        code = self.extract_code(raw_code)        return f"{self.get_stock_type(code)}|{code}"    @staticmethod    def get_workday_today():        today = datetime.today().date()        if today.weekday() == 5:            return today - timedelta(days=1)        elif today.weekday() == 6:            return today - timedelta(days=2)        else:            return today    @staticmethod    def format_date(date_obj):        return date_obj.strftime("%Y%m%d")    # 加载数据逻辑    def load_excel(self):        file_path = filedialog.askopenfilename(            title="选择Excel文件",            initialdir=self.ex_dir,           # 默认打开 Ex文件夹            filetypes=[("Excel files", "*.xlsx *.xls"), ("All files", "*.*")]        )        if not file_path:            return        # ---------- 优先方式：pd.read_excel ----------        try:            self.df_data = pd.read_excel(file_path)        except Exception as e_excel:            # 问财等网页数据读取方式 ----------            try:                # read_html 返回列表，取第一个 DataFrame                self.df_data = pd.read_html(file_path, header=0)[0]            except Exception as e_html:                messagebox.showerror(                    "错误",                    f"读取Excel失败（read_excel: {e_excel}\nread_html: {e_html})"                )                self.df_data = None                return        # ---------- 后续处理（与原来完全相同） ----------        self.current_headers = self.df_data.columns.tolist()        self.listbox.delete(0, tk.END)        for h in self.current_headers:            self.listbox.insert(tk.END, h)        self.excel_path = file_path        self.entry_path.delete(0, tk.END)        self.entry_path.insert(0, file_path)        messagebox.showinfo("成功", f"已加载 {len(self.df_data)} 行，{len(self.current_headers)} 列")    # 保存自定义数据    def preview_and_save(self):        if self.df_data is None:            messagebox.showerror("错误", "请先加载 Excel 文件")            return        selected = self.listbox.curselection()        if not selected:            messagebox.showerror("错误", "请先在列表中至少选中一个数据列")            return        # 获取选中的列名（可能多个）        selected_cols = [self.current_headers[i] for i in selected]        first_col = self.df_data.columns[0]          # 第一列为股票代码列        stock_codes_raw = self.df_data[first_col]        date_val = self.entry_date.get().strip()        if not date_val:            date_val = self.format_date(self.get_workday_today())        if not (len(date_val) == 8 and date_val.isdigit()):            messagebox.showerror("错误", "时间标签应为8位数字（YYYYMMDD），请修正日期输入框")            return        mode = self.data_type.get()        lines = []        if mode == "external":            # 遍历每一行数据            for idx, row in self.df_data.iterrows():                raw_code = row[first_col]                if pd.isna(raw_code):                    continue                # 收集选中列的非空值（经过特殊处理）                vals = []                for col in selected_cols:                    v = row[col]                    if pd.isna(v):                        continue                    # 特殊列名处理（与原逻辑保持一致）                    if isinstance(v, str):                        if "申万行业" in col:                            parts = v.split("--")                            if len(parts) >= 2:                                v = "--".join(parts[-2:])                            v = v.replace("Ⅱ", "2").replace("Ⅲ", "3")                        elif "产品名称" in col:                            v = v.replace("||", "、")                    vals.append(str(v))                # 如果所有选中列都为空，跳过该行                if not vals:                    continue                # 用空格连接所有列非空值                merged_val = " ".join(vals)                code = self.fm_stock_code(raw_code)                lines.append(f"{code}|{merged_val}\n|0.000")               if not lines:                messagebox.showwarning("警告", "没有有效数据行，未保存")                return            self.text_display.delete(1.0, tk.END)            self.text_display.insert(tk.END, "\n".join(lines))            base_name = os.path.splitext(os.path.basename(self.excel_path))[0]            # 文件名用下划线连接所有选中列名            cols_suffix = "+".join(selected_cols)            filename = f"{date_val}_{cols_suffix}_外部_{base_name}.txt"            with open(filename, "w", encoding="gb18030") as f:                f.write("\n".join(lines))            messagebox.showinfo("保存成功", f"外部数据已保存至：{filename}")        elif mode == "sequence":            # ---------- 序列数据：保持原单列逻辑，使用第一个选中的列 ----------            col_name = selected_cols[0]          # 取第一个选中的列            data_col = self.df_data[col_name]    # 定义 data_col 变量            for raw_code, val in zip(stock_codes_raw, data_col):                if pd.isna(raw_code) or pd.isna(val):                    continue                code = self.fm_stock_code(raw_code)                lines.append(f"{code}|{date_val}|{val}")            if not lines:                messagebox.showwarning("警告", "没有有效数据行，未保存")                return            self.text_display.delete(1.0, tk.END)            self.text_display.insert(tk.END, "\n".join(lines))            base_name = os.path.splitext(os.path.basename(self.excel_path))[0]            filename = f"{date_val}_{col_name}_序列_{base_name}.txt"            with open(filename, "w", encoding="gb18030") as f:                f.write("\n".join(lines))            messagebox.showinfo("保存成功", f"序列数据已保存至：{filename}")        else:            messagebox.showerror("错误", "未知模式")    # ---------- 界面布局 ----------    def create_widgets(self):        # 文件路径区域        top_frame = tk.Frame(self.parent)        top_frame.pack(fill=tk.X, padx=10, pady=5)        tk.Label(top_frame, text="Excel文件:").pack(side=tk.LEFT, padx=5)        self.entry_path = tk.Entry(top_frame, relief=tk.SUNKEN, bg="white")        self.entry_path.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5)        btn_load = tk.Button(top_frame, text="读取Excel数据", command=self.load_excel, width=12)        btn_load.pack(side=tk.LEFT, padx=5)        # 中间左右分区        middle_frame = tk.Frame(self.parent)        middle_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)        left_frame = tk.LabelFrame(middle_frame, text="选择数据列标题（外部数据按住 Ctrl或Shift键 进行多选）", padx=5, pady=5)        left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)        listbox_frame = tk.Frame(left_frame)        listbox_frame.pack(fill=tk.BOTH, expand=True)        # 修改：设置 selectmode 为 EXTENDED 支持多选        self.listbox = tk.Listbox(listbox_frame, height=15, selectmode=tk.EXTENDED)        scrollbar = tk.Scrollbar(listbox_frame, orient=tk.VERTICAL, command=self.listbox.yview)        self.listbox.configure(yscrollcommand=scrollbar.set)        self.listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)        scrollbar.pack(side=tk.RIGHT, fill=tk.Y)        right_frame = tk.LabelFrame(middle_frame, text="参数设置", padx=10, pady=10)        right_frame.pack(side=tk.RIGHT, fill=tk.Y, padx=5)        tk.Label(right_frame, text="输出模式:", anchor="w").pack(anchor=tk.W, pady=(0,5))        rb_external = tk.Radiobutton(right_frame, text="外部数据（代码|数值或文本|0.000）",                                     variable=self.data_type, value="external")        rb_external.pack(anchor=tk.W, pady=2)        rb_sequence = tk.Radiobutton(right_frame, text="序列数据（代码|日期|数值）",                                     variable=self.data_type, value="sequence")        rb_sequence.pack(anchor=tk.W, pady=2)        date_frame = tk.Frame(right_frame)        date_frame.pack(fill=tk.X, pady=10)        tk.Label(date_frame, text="设置序列数据日期 (YYYYMMDD)：").pack(side=tk.LEFT)        default_date = self.format_date(self.get_workday_today())        self.entry_date = tk.Entry(date_frame, width=12)        self.entry_date.insert(0, default_date)        self.entry_date.pack(side=tk.RIGHT)        btn_save = tk.Button(right_frame, text="预览并保存", command=self.preview_and_save,                             bg="lightblue", font=("微软雅黑", 10), width=15)        btn_save.pack(pady=15)        preview_frame = tk.LabelFrame(self.parent, text="预览结果", padx=5, pady=5)        preview_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)        self.text_display = scrolledtext.ScrolledText(preview_frame, height=12, wrap=tk.WORD)        self.text_display.pack(fill=tk.BOTH, expand=True)# ==================== 第二部分：下载东财业绩报 ====================class CJDataCollector:    """东方财富业绩报下载工具"""    def __init__(self, parent_frame):        self.parent = parent_frame        self.data = None        self.collecting = False        self.create_widgets()        self.date_options = self.generate_quarter_dates()        self.combo_date['values'] = self.date_options        if self.date_options:            self.combo_date.current(0)    @staticmethod    def _last_day_of_month(year, month):        # 使用 date_class（即 datetime.date）避免命名冲突        if month == 12:            next_month_first = date_class(year + 1, 1, 1)        else:            next_month_first = date_class(year, month + 1, 1)        return (next_month_first - timedelta(days=1)).day    def generate_quarter_dates(self):        today = date_class.today()   # 明确使用 date_class.today()        three_years_ago = today - timedelta(days=3 * 365)        quarters = []        for year in range(three_years_ago.year, today.year + 1):            for month in (3, 6, 9, 12):                last_day = self._last_day_of_month(year, month)                q_date = date_class(year, month, last_day)                if three_years_ago <= q_date <= today:                    quarters.append(q_date.strftime("%Y-%m-%d"))        return sorted(set(quarters), reverse=True)    @staticmethod    def get_market_flag(code):        if code.startswith(('60', '68')):            return '1'        if code.startswith(('30', '00')):            return '0'        if code.startswith('92'):            return '2'        return '3'    @staticmethod    def _to_percent(value):        if not value:            return ''        try:            num = float(str(value).replace('%', '').strip())            return f"{num:.2f}%"        except (ValueError, TypeError):            return str(value)    @staticmethod    def _to_wan(value):        if not value:            return ''        try:            return f"{float(value) / 10000:.2f}"        except (ValueError, TypeError):            return str(value)    def format_record(self, record):        code = record.get('SECURITY_CODE', '')        market = self.get_market_flag(code)        profit = self._to_wan(record.get('PARENT_NETPROFIT'))        revenue = self._to_wan(record.get('TOTAL_OPERATE_INCOME'))        yoy = self._to_percent(record.get('YSTZ'))        qoq = self._to_percent(record.get('SJLTZ'))        date = record.get('REPORTDATE', '')        line1 = f"{market}|{code}|{qoq} 净利润：{profit + '万'if profit else''} 营业收入{revenue + '万'if revenue else''}{yoy}{date}"        line1 = ' '.join(line1.split())        return f"{line1}\n|0.000"    def fetch_all_pages(self, date_str, progress_callback=None):        filter_str = (            f"(SECURITY_TYPE_CODE%20in%20(%22058001001%22%2C%22058001008%22))"            f"(TRADE_MARKET_CODE!%3D%22069001017%22)"            f"(REPORTDATE%3D%27{date_str}%27)"        )        base_url = (            f"https://datacenter-web.eastmoney.com/api/data/v1/get?"            f"sortColumns=UPDATE_DATE%2CSECURITY_CODE&sortTypes=-1%2C-1&"            f"reportName=RPT_LICO_FN_CPD&columns=ALL&filter={filter_str}"        )        headers = {"User-Agent": "Mozilla/5.0"}        page_size = 500        def fetch_page(page_num):            url = f"{base_url}&pageSize={page_size}&pageNumber={page_num}"            resp = requests.get(url, headers=headers, timeout=15)            resp.raise_for_status()            data = resp.json()            if not data.get("success"):                raise Exception(data.get('message', '接口错误'))            result = data.get("result", {})            return result.get("data", []), result.get("pages", 0)        try:            first_data, total_pages = fetch_page(1)            all_data = first_data[:]            if progress_callback:                progress_callback(f"总页数: {total_pages}, 本页 {len(first_data)} 条")            if total_pages == 0:                if progress_callback:                    progress_callback("无数据")                return None            for page in range(2, total_pages + 1):                page_data, _ = fetch_page(page)                if not page_data:                    break                all_data.extend(page_data)                if progress_callback:                    progress_callback(f"已获取第 {page} 页，累计 {len(all_data)} 条")        except Exception as e:            if progress_callback:                progress_callback(f"请求失败: {e}")            return None        fields = ["SECURITY_CODE", "SECURITY_NAME_ABBR", "REPORTDATE",                  "TOTAL_OPERATE_INCOME", "PARENT_NETPROFIT", "YSTZ", "SJLTZ"]        cleaned = []        for rec in all_data:            item = {k: rec.get(k) for k in fields}            if item.get("REPORTDATE"):                item["REPORTDATE"] = item["REPORTDATE"].split()[0]            cleaned.append(item)        return cleaned    def save_to_txt(self, data, date_str, save_dir=None):        if save_dir is None:            save_dir = os.getcwd()        filepath = os.path.join(save_dir, f"{date_str}业绩报.txt")        with open(filepath, "w", encoding="gb18030") as f:            for rec in data:                f.write(self.format_record(rec) + "\n")        return filepath    def create_widgets(self):        top = ttk.Frame(self.parent, padding=10)        top.pack(fill=tk.X)        ttk.Label(top, text="报告日期：", font=("Arial", 12)).pack(side=tk.LEFT, padx=5)        self.combo_date = ttk.Combobox(top, width=15, font=("Arial", 12), state="readonly")        self.combo_date.pack(side=tk.LEFT, padx=5)        self.collect_btn = ttk.Button(top, text="下载业绩报", command=self.start_collect)        self.collect_btn.pack(side=tk.LEFT, padx=30)        self.save_btn = ttk.Button(top, text="保存业绩报自定义外部数据", command=self.save_data, state=tk.DISABLED)        self.save_btn.pack(side=tk.LEFT, padx=5)        preview_frame = ttk.LabelFrame(self.parent, text="数据预览（前500条，单位：万元，同比/环比已格式化）", padding=5)        preview_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)        cols = ("代码", "名称", "报告日期", "营业总收入(万元)", "营收同比增长%", "净利润(万元)", "净利润环比增长%")        self.tree = ttk.Treeview(preview_frame, columns=cols, show="headings", height=8)        for col in cols:            self.tree.heading(col, text=col)            self.tree.column(col, width=120, anchor="center")        vsb = ttk.Scrollbar(preview_frame, orient="vertical", command=self.tree.yview)        self.tree.configure(yscrollcommand=vsb.set)        self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)        vsb.pack(side=tk.RIGHT, fill=tk.Y)        log_frame = ttk.LabelFrame(self.parent, text="下载日志", padding=5)        log_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)        self.log_text = scrolledtext.ScrolledText(log_frame, wrap=tk.WORD, height=15, font=("Consolas", 10))        self.log_text.pack(fill=tk.BOTH, expand=True)    def log(self, msg):        self.log_text.insert(tk.END, msg + "\n")        self.log_text.see(tk.END)        self.parent.update_idletasks()    def start_collect(self):        if self.collecting:            self.log("下载中，请勿重复点击")            return        date = self.combo_date.get()        if not date:            messagebox.showwarning("警告", "请选择报告日期")            return        self.data = None        self.save_btn.config(state=tk.DISABLED)        for row in self.tree.get_children():            self.tree.delete(row)        self.log_text.delete(1.0, tk.END)        self.collecting = True        self.collect_btn.config(state=tk.DISABLED, text="下载中...")        threading.Thread(target=self._collect_thread, args=(date,), daemon=True).start()    def _collect_thread(self, date_str):        def progress(msg):            self.parent.after(0, lambda: self.log(msg))        result = self.fetch_all_pages(date_str, progress)        self.parent.after(0, lambda: self._on_finished(result))    def _on_finished(self, data):        self.collecting = False        self.collect_btn.config(state=tk.NORMAL, text="下载业绩报")        if not data:            self.log("下载失败")            return        self.data = data        self.log(f"下载完成，共 {len(data)} 条")        self.save_btn.config(state=tk.NORMAL)        for row in self.tree.get_children():            self.tree.delete(row)        for rec in data[:500]:            self.tree.insert("", tk.END, values=(                rec.get("SECURITY_CODE", ""),                rec.get("SECURITY_NAME_ABBR", ""),                rec.get("REPORTDATE", ""),                self._to_wan(rec.get("TOTAL_OPERATE_INCOME")),                self._to_percent(rec.get("YSTZ")),                self._to_wan(rec.get("PARENT_NETPROFIT")),                self._to_percent(rec.get("SJLTZ"))            ))    def save_data(self):        if not self.data:            messagebox.showwarning("警告", "无数据可保存")            return        try:            path = self.save_to_txt(self.data, self.combo_date.get())            self.log(f"已保存: {path}")            messagebox.showinfo("保存成功", f"文件位于:\n{path}")        except Exception as e:            messagebox.showerror("保存失败", str(e))# ==================== 合并主窗口 ====================class MergedApp:    def __init__(self, root):        self.root = root        self.root.title("Tdx自定义数据工具集 —— 小红牛微信公众号：gxzfp888")        self.root.geometry("1100x750")        # 设置图标（如果存在）        base_path = os.path.abspath(os.path.dirname(__file__))        ico_path = os.path.join(base_path, "resources", "采集.ico")        if os.path.exists(ico_path):            try:                self.root.iconbitmap(ico_path)            except:                pass        # 创建选项卡        self.notebook = ttk.Notebook(root)        self.notebook.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)        # 第一个选项卡：Excel转通达信        self.tab1 = ttk.Frame(self.notebook)        self.notebook.add(self.tab1, text=" Excel转通达信自定义数据 ")        self.excel_tool = StockDataProcessor(self.tab1)        # 第二个选项卡：下载业绩报        self.tab2 = ttk.Frame(self.notebook)        self.notebook.add(self.tab2, text=" 下载东财业绩报 ")        self.cj_tool = CJDataCollector(self.tab2)# 启动应用if __name__ == "__main__":    root = tk.Tk()    app = MergedApp(root)    root.mainloop()