基于Python+DeepSeek的Excel数据处理程序

基于ds的excel数据处理程序（批量处理地址数据）

程序功能就是把excel表格中某一列的各种公司名称什么的东西，通过ai找到他们所属的省份，再写入到文件里

# ---------------------------- 导入库部分 ----------------------------
import xlrd  # 用于读取Excel文件，但是这个库只能读取.xls格式的文件，需要下载旧版本的库才能读取xlsx格式的文件
from openai import OpenAI  # DeepSeek API调用接口
import openpyxl  # 用于写入新版Excel(.xlsx)文件
import json  # 处理JSON格式数据
import time  # 时间控制(重试间隔/退出等待)


# ---------------------------- 全局配置部分 ----------------------------
api_key = "阿巴阿巴歪比巴卜"  # DeepSeek API密钥(⚠️注意：实际使用时应避免硬编码)


# ---------------------------- 核心函数定义 ----------------------------
def shuju_huoqu():
    """
    数据获取函数
    功能：读取Excel文件路径、选择表单、提取指定列数据
    返回：原始数据列表、起始行号、新文件路径、表单索引、结束行号、原始文件路径
    """
    # 处理文件路径输入(自动去除引号)
    old_file_dizhi = input("输入文件地址：").strip('"')
    # 构建新文件路径(与原文件同目录)
    index = old_file_dizhi.rfind("\\")
    new_file_dizhi = old_file_dizhi[:index] + "\\处理后的表格.xlsx"

    # 打开Excel工作簿
    book = xlrd.open_workbook(old_file_dizhi)
    print(f" - 该文件包含表单数量为：{book.nsheets}")
    print(f" - 各个表单的名称分别为：")
    # 打印所有表单名称
    for i in book.sheet_names():
        print("  " + i)
    print("\n")

    # 用户选择表单(转换为0-based索引)
    biaodan_shu = int(input("选择表单（从上到下依次为1，2...)：")) - 1
    biaodan = book.sheet_by_index(biaodan_shu)
    print(f" - 表单名：{biaodan.name}    包含 {biaodan.nrows} 行 {biaodan.ncols} 列 ")

    # 打印表头信息
    print(" - 该表单表头由左到右依次为：")
    i = 0
    while (i < biaodan.ncols):
        k = i + 1
        print(f"  {k}." + str(biaodan.cell_value(rowx=0, colx=i)))
        i += 1

    # 用户选择数据列(转换为0-based索引)
    print(" ")
    col = int(input("数据位于第几列（从左到右依次为1，2...）：")) - 1

    # 统计有效数据行数(非空值计数)
    count = 0
    for row in range(biaodan.nrows):
        cell_value = str(biaodan.cell_value(row, col))
        if cell_value.strip() != "":
            count += 1

    # 显示前5项数据样例
    print(f" - 该列数据共 {count} 行，前五项元素为：")
    i = 0
    while (i < 5):
        k = i + 1
        print(f"  {k}." + str(biaodan.cell_value(rowx=i, colx=col)))
        i += 1

    # 获取数据起始行(转换为0-based索引)
    print(" ")
    biaotou = int(input("数据从第几行开始（除表头）：")) - 1
    print(f" - 数据从第 {biaotou + 1} 行到第 {count} 行")

    # 提取数据到列表
    print("开始获取数据...")
    shuju = []
    for i in range(biaotou, count):
        value = str(biaodan.cell_value(rowx=i, colx=col))
        shuju.append(value)

    print(" - 获取的数据：")
    print(str(shuju))
    return shuju, biaotou, new_file_dizhi, biaodan_shu, count - 1, old_file_dizhi


def shuju_chuli(shuju):
    """
    数据去重函数
    参数：原始数据列表
    返回：保持顺序的去重后列表
    """
    c_shuju = []
    for j in shuju:
        if j not in c_shuju:  # 通过遍历实现顺序保持
            c_shuju.append(j)
    print("数据处理完成，处理后的数据：")
    print(c_shuju)
    return c_shuju


def diaoyong_deepseek(c_shuju):
    """
    DeepSeek API调用函数
    参数：去重后的数据列表
    返回：字典{原始数据: 省份}
    """
    j_shuju = []  # 分批存储数据
    d_shuju = {}  # 最终结果字典
    total = len(c_shuju)

    # 分批处理(每批300条)
    for i in range(0, total, 250):
        batch = c_shuju[i:i + 250]
        print(f"正在发送第{i // 250 + 1}批数据...")

        # 带重试机制的API调用
        while True:
            try:
                client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")
                response = client.chat.completions.create(
                    model="deepseek-reasoner",
                    messages=[
                        {
                            "role": "system",
                            "content": """
                            处理规则：
                            1. 输入列表元素→匹配省份
                            2. 无法匹配→"***所在地址未知***"
                            3. 输出纯JSON格式
                            """
                        },
                        {"role": "user", "content": str(batch) + "返回结果应仅包含一个json格式的数据集"},
                    ],
                    temperature=0,  # 确定性输出
                    stream=False,  # 非流式响应
                    max_tokens=8000  # 最大token限制
                )
            except Exception as e:
                print("API调用失败，正在重试...")
                time.sleep(1)
                continue
            else:
                # 解析返回内容
                content = response.choices[0].message.content
                # 提取JSON部分(兼容可能的附加说明)
                start = content.find('{')
                end = content.rfind('}') + 1
                if start != -1 and end != 0:
                    result = json.loads(content[start:end])
                    d_shuju.update(result)
                    print("当前批次处理结果：", result)
                break
            finally:
                client.close()
    return d_shuju


def shuju_luru(d_shuju, shuju, kaitou_hang, new_file_dizhi, biaodan_shu, jiewei_hang, old_file_dizhi):
    """
    数据写入函数
    功能：将省份信息写入Excel指定列
    """
    # 生成映射列表(保持原始顺序)
    liebiao = [d_shuju.get(item, "***所在地址未知***") for item in shuju]

    # 打开Excel文件
    excel = openpyxl.load_workbook(old_file_dizhi)
    sheet = excel.worksheets[biaodan_shu]

    # 用户选择写入列(转换为1-based索引)
    lie = int(input("选择数据将要写入的列数（从左到右依次为1，2...："))

    # 写入数据到指定列
    for idx, value in enumerate(liebiao):
        sheet.cell(row=kaitou_hang + idx + 1, column=lie, value=value)

    # 带异常处理的保存机制
    while True:
        try:
            excel.save(new_file_dizhi)
            break
        except Exception as e:
            print(f"保存失败：{e}（可能文件被占用，请关闭后重试）")
            time.sleep(5)

    print("保存成功，新文件与原文件位于同一目录下")



# ---------------------------- 主程序执行流程 ----------------------------
if __name__ == "__main__":
    # 阶段1：数据获取
    shuju, kaitou_hang, new_file_dizhi, biaodan_shu, jiewei_hang, old_file_dizhi = shuju_huoqu()

    # 阶段2：数据预处理
    c_shuju = shuju_chuli(shuju)

    # 阶段3：调用API处理数据
    d_shuju = diaoyong_deepseek(c_shuju)

    # 阶段4：结果写入Excel
    shuju_luru(d_shuju, shuju, kaitou_hang, new_file_dizhi, biaodan_shu, jiewei_hang, old_file_dizhi)

大概就是把excel数据取出来，去重，然后一批一批发给deepseek处理，返回数据强制为json格式，然后在本地建立地址和所属省份的映射，数据处理，然后再把处理好的数据写回去。