hand
_1_5_89
4
返回栏目
1k
9k
1k
1k
5k
1k
1k
1k
1k
3k
2k
1k
0.8k
2k
3k
1k
1k
0.7k
0.9k
1k
0.6k
0.4k
0.4k
0.3k
3k
2k
9k
0.4k
0.4k
0.8k
0.5k
3k
5k
1k
2k
2k
3k
5k
1k
1k
0.4k
0.5k
0.4k
0.6k
0.7k
1k
0.4k
0.3k
4k
0.5k
0k
0.3k
0k
0.2k
0.2k
0.3k
0.9k
0.9k
0.1k
0.9k
0.9k
1k
0.5k
6k
0.3k
0.4k
0.7k
0.6k
8k
3k
1k
1k
1k
1k
0k
2k
1k
1k
0.2k
5k
4k
5k
0.4k
0.8k
1k
1k
1k
0.1k
2k
1k
2k
6k
0k
2k
7k
1k
5k
2k
3k
1k
0k
1k
0.9k
0.4k
0.2k
1k
3k
4k
1k
1k
1k
2k
3k
0.7k
0.3k
0.5k
0.6k
1k
0.9k
3k
0.3k
4k
返回python栏目
作者:
贺及楼
成为作者
更新日期:2024-05-30 12:12:49
3个类:
一个是数据库类 = SQL()
一个是word文档类 = WORD()
一个是逻辑类 = Main()
先用Main().list()方法找到要跑的表
也可以全部表复制进去
如果一条记录都没有的,不写入表
# -*- coding: utf-8 -*-
import json
from textwrap import indent
import pymysql
from sqlalchemy import create_engine
import docx
import pandas as pd
import numpy as np
class SQL():
def __init__(self):
self.ip = "ip"
self.port = "3306"
self.port2 = 3306
self.username = "root"
self.secret = "secret"
self.dbname = "dbname"
# self.tablename = "tablename"
def connect(self):
url = 'mysql+pymysql://'+self.username+':'+self.secret+'@'+self.ip+':'+self.port+'/'+self.dbname+'?charset=utf8mb4'
print(url)
connect = create_engine(url)
return connect
def connect2(self):
db = pymysql.connect(host=self.ip, user=self.username, password=self.secret, port=self.port2,database=self.dbname)
cursor = db.cursor() # 获得指针
return cursor
def select(self):
cursor = self.connect2()
sql = 'select * from dbname.tablename;'
cursor.execute(sql) # 执行
return cursor
def all_tables(self):
cursor = self.connect2()
sql = "use "+self.dbname+";"
cursor.execute(sql) # 执行
sql = "show tables;"
cursor.execute(sql) # 执行
return list(cursor)
def select_desc(self, table_name):
cursor = self.connect2()
sql = "desc "+table_name+";"
cursor.execute(sql) # 执行
return cursor
def select_one(self, table_name):
try:
cursor = self.connect2()
sql = "select * from "+table_name+" where id = 1;"
cursor.execute(sql) # 执行
data = cursor.fetchall()
cols = cursor.description
data = list(data)
print('cols')
print(cols)
print('cols')
return data, cols
except:
return [], []
class WORD():
def __init__(self):
print("")
def doc(self):
doc = docx.Document()
return doc
def write_table(self, doc, df):
t = doc.add_table(df.shape[0]+1, df.shape[1])
# print(df)
# add the header rows.
for j in range(df.shape[-1]):
t.cell(0,j).text = str(df.columns[j])
# add the rest of the data frame
for i in range(df.shape[0]):
for j in range(df.shape[-1]):
t.cell(i+1,j).text = str(df.values[i,j])
return
# save the doc
def write_dict(self, doc, data):
data = data.to_dict('records')
data_str = json.dumps(data,indent=2, default=str,ensure_ascii=False)
p3 = doc.add_paragraph()
p3.add_run(data_str)
def save(self, doc):
doc.save('./test.docx')
class Main():
def __init__(self):
self.table_list = ['user','student'] # 修改需要生成的表
def list(self):
table_list = SQL().all_tables()
table_list_T = []
for i in table_list:
table_list_T.append(i[0])
return table_list_T
def run(self):
Sql = SQL()
Word = WORD()
doc = Word.doc() # 新文档
for i in self.table_list:
desc_data = Sql.select_desc(i)
desc_data = pd.DataFrame(data=desc_data)
one_data, cols = Sql.select_one(i)
col = []
for c in cols:
col.append(c[0])
one_data = pd.DataFrame(data=one_data, columns=col)
print(one_data)
# if one_data==[]:
if one_data.empty:
continue
desc_data["解释"] = ""
desc_data.columns = ["名字","类型","必填","主键","默认","附加","解释"]
p3 = doc.add_paragraph()
p3.add_run(i)
Word.write_table(doc, desc_data)
Word.write_dict(doc, one_data)
Word.save(doc)
if __name__=="__main__":
# 全部列表,然后要贴到Main()的self.table_list
# table_list = Main().list()
# print(table_list)
Main().run()
python
整章节共122节
快分享给你的小伙伴吧 ~