添加文件读写函数

Author: jerrychen1990

.gitignore

@@ -134,3 +134,4 @@ upload_package.sh
 
 test.ipynb
 tests/agi-tools.code-workspace
+.DS_Store

setup.py

@@ -14,7 +14,7 @@
 
 from setuptools import find_packages, setup
 
-from snippets.utils import get_latest_version, get_next_version
+from snippets.utils import get_latest_version, get_next_version, read2list
 
 REQ = [
     "tqdm",
@@ -23,6 +23,11 @@
     "click"
 ]
 
+def get_install_req():
+    req = read2list("requirements.txt")
+    return req
+
+
 
 if __name__ == "__main__":
     name = "python-snippets"
@@ -32,7 +37,8 @@
         latest_version = get_latest_version(name)
         version = get_next_version(latest_version)
     print(f"version: {version}")
-
+    install_req = get_install_req()
+    print(f"install_req: {install_req}")
     setup(
         name=name,
         version=version,

snippets/mixin.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+# -*- encoding: utf-8 -*-
+'''
+@Time    :   2023/10/16 14:09:07
+@Author  :   ChenHao
+@Contact :   jerrychen1990@gmail.com
+'''
+
+from utils import jload
+from typing import Union
+
+
+class ConfigMixin:
+    @classmethod
+    def from_config(cls, config: Union[dict, str]):
+        if isinstance(config, str):
+            if config.endswith(".json"):
+                config = jload(config)
+            else:
+                raise ValueError(f"{config} is not a valid config file")
+        instance = cls(**config)
+        return instance

snippets/utils.py

@@ -152,8 +152,6 @@ def get_gen(f):
     return list(gen)
 
 # table类的文件转化为list of dict
-
-
 def table2json(path):
     if path.endswith("csv"):
         df = pd.read_csv(path)
@@ -163,9 +161,24 @@ def table2json(path):
     records = df.to_dict(orient="records")
     return records
 
-# 一行一行地读取文件内容
+
+# 将list数据存储成table格式
+def dump2table(data, path):
+    if isinstance(data, list):
+        data = pd.DataFrame.from_records(data)
+    assert isinstance(data, pd.DataFrame)
+    df = data
+    
+    if path.endswith(".csv"):
+        df.to_csv(path, index=False)
+    elif path.endswith(".xlsx"):
+        df.to_excel(path, index=False)
+    else:
+        raise Exception(f"Unknown file format: {path}")
+    
 
 
+# 一行一行地读取文件内容
 def load_lines(fp, return_generator=False):
     if isinstance(fp, str):
         fp = open(fp, mode="r", encoding="utf8")
@@ -175,7 +188,7 @@ def load_lines(fp, return_generator=False):
             return (e.strip() for e in lines if e)
         return [e.strip() for e in lines if e]
 
-
+# 根据后缀名读取list数据
 def read2list(file_path: str, **kwargs) -> List[Union[str, dict]]:
     name, surfix = split_surfix(file_path)
     if surfix == ".json":
@@ -189,6 +202,22 @@ def read2list(file_path: str, **kwargs) -> List[Union[str, dict]]:
     else:
         logger.warn(f"unkown surfix:{surfix}, read as txt")
         return load_lines(file_path, **kwargs)
+    
+
+# 将list数据按照后缀名格式dump到文件
+def dump_list(data:List, file_path: str, **kwargs):
+    name, surfix = split_surfix(file_path)
+    if surfix == ".json":
+        return jdump(data, file_path, **kwargs)
+    if surfix == ".jsonl":
+        return jdump_lines(data, file_path, **kwargs)
+    if surfix in ["xlsx", "csv"]:
+        return dump2table(data, file_path)
+    if surfix in ["txt"]:
+        return dump_lines(file_path, **kwargs)
+    else:
+        logger.warn(f"unkown surfix:{surfix}, dump as txt")
+        return load_lines(file_path, **kwargs)
 
 
 # 递归将obj中的float做精度截断