Python代码实例

Python快速入门&慢速出门 PaddlePaddle_OCR

调用COM

# import moudles
from win32com.client import Dispatch
 
import time;
import ctypes
from ctypes import *
LD = ctypes.windll.LoadLibrary;
#参数自己设置成FreeCom.dll的路径
freeCOM = LD(R"E:\project\op\bin\x86\tools.dll");
#参数自己设置成op_x86.dll的路径
 
ret = freeCOM.setupA(bytes(R"E:\project\op\bin\x86\op_x86.dll",encoding="utf-8"));
 
print("setupA:{}".format(ret));
# create op instance
op = Dispatch("op.opsoft");
print(op.Ver())
op.MoveTo(30,30)

Excel / JSON

JSON 处理

https://www.runoob.com/python3/python3-json.html

import json
# 转为字符串, 允许 python表达式
str = json.dumps(['foo', {'bar': ('baz', None, 1.0, 2)}], ensure_ascii=False, allow_nan=False)
# print(str)
# 转为对象
obj = json.loads('["foo", {"bar":["baz", null, 1.0, 2]}]')
print(obj[0])
print(obj[1]["bar"])
 
 
 
def get_json_from_file(input_file ):
    with open(input_file, 'r', encoding='utf-8') as f:
        json_data = json.loads(f.read())
        return json_data
        
def get_json_from_file_bom(input_file ):
	# windows的 BOM问题
	with open(input_file, 'r', encoding='utf-8-sig') as f:
		json_data = json.loads(f.read())
		return json_data
		
def save_json_to_file(ouput_file_path, json_element):
    content = json.dumps(json_element, ensure_ascii=False, allow_nan=False)
    with open(ouput_file_path,mode='w',encoding="utf-8") as output_file :
        output_file.write(content)
 

包含 NaN 的问题 NaN是一个特例，它在 Python 里被转为字符串”NaN”，并且可以被json.loads识别。 在 Javascript 中，JSON.stringify(NaN)被转为”null”，并且无法转回。其原因是NaN不在 json 标准内。

json.dumps 有一个 allow_nan 参数，默认为 True。NaN，Infinity 不是JSON的一部分，如果想json.dumps的结果中不包括 NaN，可以将 allow_nan 设置为 false，但是当 json.dumps 的参数中包含NaN时，会抛出 ValueError 的异常。

字符串IO 处理

input_file = 'E:\\content-for-work\\202310深汕建设项目\\合作社列表.json'
def read_from_file(path):
	with open(path, 'r', encoding='utf-8') as f:
		data = f.read()
		return data
 
 
def save_to_file(path, content):
    with open(path, mode='w', encoding="utf-8") as output_file :
        output_file.write(content)
 
 

Excel to JSON

import pandas as pd
import json
 
# 读取Excel文件
excel_file = "E:\\content-for-work\\202310深汕建设项目\\2023-深汕-合作社002.xlsx"
df = pd.read_excel(excel_file)
# 将所有NaN数据替换为特定值（例如，"N/A"）
df = df.fillna("")
# 将数据框（DataFrame）转换为字典列表
data = df.to_dict(orient="records")
# 将字典列表保存为JSON文件
json_file = excel_file + ".json"
with open(json_file, "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, allow_nan=False)
 
print(f"Excel文件已成功转换为JSON文件 {json_file}")
 

需安装 pip install pandas

JSON to Excel

import pandas as pd
import json
# 从JSON文件加载数据
input_file='E:\\CONTENT-FOR-WORK\\2023-07三普\\全程溯源API.JSON'
json_data = pd.read_json(input_file)
# 创建一个Pandas DataFrame
df = pd.DataFrame(json_data)
# 将DataFrame保存为Excel文件
df.to_excel(input_file+'.xlsx', index=False)
print("the end..")

Excel 插图+JSON

import pandas as pd
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
from openpyxl.drawing.image import Image
import os
 
def create_excel_with_images(data, output_file, image_columns, value_columns):
    # 将 JSON 数据转换为 DataFrame
    df = pd.DataFrame(data)
    # 创建一个新的 Excel 工作簿
    wb = Workbook()
    ws = wb.active
    # 写入数据到 Excel
    for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=True), 1):
        ws.append(row)
    # 设置每行的高度
    for row_index in range(1, len(data) + 2):
        ws.row_dimensions[row_index].height = 80
        ws.row_dimensions[row_index].width = 200
    # 处理全部列（图片列和值列）
    all_columns = list(data[0].keys())# image_columns + value_columns
    print("all_columns = ", all_columns)
    for idx, row in enumerate(data, start=2):
        for val in all_columns:
            col_letter = chr(65 + all_columns.index(val))
            
            if val in image_columns:
                # 处理图片列
                img_data = row[val]  # Assuming image column contains relative image path
                if (img_data == ""):
                    print("没有图片数据 样点编号: ", row["样点编号"])
                    continue
                img_path = g_rootpath+img_data
                img = Image(img_path)
                img.width = 100  # 设置图片宽度
                img.height = 50  # 设置图片高度
                ws.add_image(img, f"{col_letter}{idx}")
# 示例数据
json_data = [
    {"姓名": "xxx", "头像": "E:\\content-for-work\\200004-6463e75a2960.jpg", "值1": "abc", "值2": 123},
    {"姓名": "xxx2", "头像": "E:\\content-for-work\\20230-6463e75a2960.jpg", "值1": "def", "值2": 456}
]
# 定义图片列和值列
image_columns = ["头像"]
value_columns = ["姓名", "值1" ]
 
create_excel_with_images(json_data, "18998910542_用户_output.xlsx", image_columns, value_columns)

递归遍历文件

def recursive_all_file(directory):
    file_paths = []
    for root, dirs, files in os.walk(directory):
        for name in files:
            if name.endswith(".txt"):
                file_path = os.path.join(root, name)
                # print(os.path.join(root, name))
                file_paths.append(file_path)
    return file_paths 

HTTP 请求 处理

proxies = { "http": None, "https": None}
params = {
    'key': '',
    'address': '海丰县赤石镇新城村委下坡村南埔',
}
response = requests.get("https://restapi.amap.com/v3/geocode/geo", params,  proxies=proxies, verify=False)
res_json = response.json()
if (res_json["info"] == 'OK'):
    res_json["geocodes"]
    print( res_json )

下载文件

def download_file(url, output_file):
    output_directory = os.path.dirname(output_path)
    if not os.path.exists(output_directory):
        os.makedirs(output_directory)
    r = requests.get(url, stream=True)
    with open(output_file, "wb") as f:
        for bl in r.iter_content(chunk_size=1024):
            if bl:
                f.write(bl)

地址 查 经纬度(高德)

import json
import requests
import time
# 禁用ssl错误警告
requests.packages.urllib3.disable_warnings()
 
def get_json_data(input_file ):
    with open(input_file, 'r', encoding='utf-8') as f:
        json_data = json.loads(f.read())
        return json_data
######################################################## 
proxies = { "http": None, "https": None}
params = {
    'key': 'cda3228db987ce77f96e8947f3681910',
    'address': '海丰县赤石镇新城村委下坡村南埔',
}
#########
input_file_path = 'E:\\content-for-work\\202310深汕建设项目\\2023-深汕-合作社002.xlsx.json'
in_json_data = get_json_data(input_file_path)
# address = json_data[0]["address"]
# print(address)
for element in in_json_data: 
    params['address'] = element['address']
    time.sleep(0.055)# 避免 高频请求
    response = requests.get("https://restapi.amap.com/v3/geocode/geo", params,  proxies=proxies, verify=False)
    res_json = response.json()
    if (res_json["info"] == 'OK'):
        element["geocodes"] = res_json["geocodes"]
        print("查询成功: {0}".format(element["id"]) )
    else:
        element["geocodes"] = "查询错误"
        print("查询失败: {0}, 响应: ".format(element["id"] , res_json) )
print("最终结果", in_json_data)
output_file_path  = 'E:\\content-for-work\\202310深汕建设项目\\合作社列表_查询.json'
with open(output_file_path,mode='w',encoding="utf-8") as output_file :
    out_str = json.dumps(in_json_data, ensure_ascii=False)
    output_file.write(out_str)
 

apngasm 库

https://github.com/laggykiller/apngasm-python

Install

pip install apngasm-python --trusted-host=pypi.python.org 

Optionally, you can also install Pillow and numpy

pip install Pillow numpy

 
from apngasm_python.apngasm import APNGAsmBinder
import numpy as np
from PIL import Image
import os
 
def mian():
    apngasm = APNGAsmBinder()
    # From file
    for file_name in sorted(os.listdir('frames')):
        # To adjust frame duration, set delay_num and delay_den
        # The frame duration will be (delay_num / delay_den) seconds
        apngasm.add_frame_from_file(file_path=os.path.join('frames', file_name), delay_num=100, delay_den=1000)
        
    # Default value of loops is 0, which is infinite looping of APNG animation
    # This sets the APNG animation to loop for 3 times before stopping
    apngasm.set_loops(3)
    apng.assemble('result-from-file.apng')
    apngasm.reset()
 
    # From Pillow
    for file_name in sorted(os.listdir('frames')):
        image = Image.open(os.path.join('frames', file_name)).convert('RGBA')
        frame = apngasm.add_frame_from_pillow(image, delay_num=50, delay_den=1000)
    apngasm.assemble('result-from-pillow.apng')
    apngasm.reset()
 
    # Disassemble and get pillow image of one frame
    # You can use with statement to avoid calling reset()
    with APNGAsmBinder() as apng:
        frames = apng.disassemble_as_pillow('input/ball.apng')
        frame = frames[0]
        frame.save('output/ball0.png')
 
    # Disassemble all APNG into PNGs
    apngasm.save_pngs('output')
 
if __name__ == '__main__':
    mian()

数据库连接 - PyMySQL 驱动

https://www.runoob.com/python3/python3-mysql.html

安装 pip3 install PyMySQL

pip install PyMySQL—trusted-host https://pypi.org

连接

def get_db_connect(hostname,username, password, db):
   #连接数据库
    db=pymysql.connect(host=hostname,user=username,password=password, db=db,charset='utf8')
    return db
 
def main():
    hostname = "192.168.50.248"
    username = "user_dev"
    password = "dev-sny.com"
    db="nyfwzxjs-admin"
    connent = get_db_connect(hostname,username,password, db)
    cursor = connent.cursor()#创建一个游标对象
    try:
	    # 执行更新代码
        update_test(cursor)
    except Exception as e:
        print(e)
        connent.rollback()
    finally:
        connent.commit()
        cursor.close()
        connent.close()

创建表

cursor=db.cursor() # 创建一个游标对象
cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
# 使用预处理语句创建表
sql = """CREATE TABLE EMPLOYEE (
         FIRST_NAME  CHAR(20) NOT NULL,
         LAST_NAME  CHAR(20),
         AGE INT,  
         SEX CHAR(1),
         INCOME FLOAT )"""
 
cursor.execute(sql)
 

插入

# SQL 插入语句
sql = """INSERT INTO EMPLOYEE(FIRST_NAME,
         LAST_NAME, AGE, SEX, INCOME)
         VALUES ('Mac', 'Mohan', 20, 'M', 2000)"""
try:
   # 执行sql语句
   cursor.execute(sql)
   # 提交到数据库执行
   db.commit()
except:
   # 如果发生错误则回滚
   db.rollback()
# 关闭数据库连接
db.close()

更新

# SQL 更新语句
sql = "UPDATE EMPLOYEE SET AGE = AGE + 1 WHERE SEX = '%c'" % ('M')
try:
   # 执行SQL语句
   cursor.execute(sql)
   # 提交到数据库执行
   db.commit()
except:
   # 发生错误时回滚
   db.rollback()
# 关闭数据库连接
db.close()

查询

# SQL 查询语句
sql = "SELECT * FROM EMPLOYEE \
       WHERE INCOME > %s" % (1000)
try:
   # 执行SQL语句
   cursor.execute(sql)
   # 获取所有记录列表
   results = cursor.fetchall()
   for row in results:
      fname = row[0]
      lname = row[1]
      age = row[2]
      sex = row[3]
      income = row[4]
       # 打印结果
      print ("fname=%s,lname=%s,age=%s,sex=%s,income=%s" % \
             (fname, lname, age, sex, income ))
except:
   print ("Error: unable to fetch data")
 

close():关闭此游标对象
fetchone():得到结果集的下一行
fetchmany([size = cursor.arraysize]):得到结果集的下几行
fetchall():得到结果集中剩下的所有行
excute(sql[, args]):执行一个数据库查询或命令
excutemany(sql, args):执行多个数据库查询或命令