# -*- coding: utf-8 -*-"""python 自定义库测试用途:验证自定义模块在 QMT 受限环境下的文件读写拦截情况口径日期:2026-10-13受限说明(对应文档 2.1 / 2.2): - 写入:to_excel / to_csv / to_json / to_parquet -> 受限制 - 读取:read_excel / read_csv / json.load / read_parquet -> 受限制 - 纯内存 DataFrame 计算 -> 不受影响"""import osimport jsonimport pandas as pdclass xg_test: def __init__(self, base_dir=r"C:\Users\lx6\Desktop\xg_test"): """ 初始化:统一管理测试数据目录与文件路径 """ self.base_dir = base_dir # 统一维护所有文件路径,供各方法共用 self.file_paths = { "excel": os.path.join(self.base_dir, "test.xlsx"), "csv": os.path.join(self.base_dir, "test.csv"), "json": os.path.join(self.base_dir, "test.json"), "parquet": os.path.join(self.base_dir, "test.parquet"), } # 确保目录存在 if not os.path.exists(self.base_dir): os.makedirs(self.base_dir) # -------------------------------------------------------- # 生成测试数据 # -------------------------------------------------------- def get_test_data(self): """ 生成测试数据,并写入 Excel / CSV / JSON / Parquet 注意:写入操作在受限环境下会被拦截 """ df = pd.DataFrame() df["测试"] = [1, 2, 3, 4, 5, 6, 7, 8] df.to_excel(self.file_paths["excel"], index=False) df.to_csv(self.file_paths["csv"], index=False, encoding="utf-8-sig") df.to_json(self.file_paths["json"], force_ascii=False, orient="records") df.to_parquet(self.file_paths["parquet"], index=False) print("[生成] 测试数据目录:%s" % self.base_dir) for k, v in self.file_paths.items(): print(" - %-8s : %s" % (k, v)) return self.file_paths # -------------------------------------------------------- # 纯内存返回(不受限) # -------------------------------------------------------- def test(self): """ 正常的数据返回(纯内存计算) 受限环境下应保持可用 """ df = pd.DataFrame() df["测试"] = [1, 2, 3, 4, 5, 6, 7, 8] return df # -------------------------------------------------------- # 读取方法(受限) # -------------------------------------------------------- def read_json(self): """ 读 JSON(文件读取,受限) """ with open(self.file_paths["json"], "r", encoding="utf-8") as f: obj = json.load(f) return obj def read_excel(self): """ 读 Excel(文件读取,受限) """ df = pd.read_excel(self.file_paths["excel"]) return df def read_csv(self): """ 读 CSV(文件读取,受限) """ df = pd.read_csv(self.file_paths["csv"]) return df def read_parquet(self): """ 读 Parquet(文件读取,受限) """ df = pd.read_parquet(self.file_paths["parquet"]) return df # -------------------------------------------------------- # 路径查看 # -------------------------------------------------------- def show_paths(self): """ 打印当前所有文件路径,便于排查 """ for k, v in self.file_paths.items(): print("%-8s : %s" % (k, v)) return self.file_pathsif __name__ == "__main__": test = xg_test() # 查看路径 test.show_paths() # 生成测试数据 test.get_test_data() # 依次读取测试 print("\n--- read_excel ---") print(test.read_excel()) print("\n--- read_csv ---") print(test.read_csv()) print("\n--- read_json ---") print(test.read_json()) print("\n--- read_parquet ---") print(test.read_parquet()) print("\n--- 纯内存 test() ---") print(test.test())
发表评论
发表评论: