Python¶

import datetime
import locale
import smtplib
from email.mime.image import MIMEImage
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from email.utils import formataddr

locale.setlocale(locale.LC_ALL, 'zh_CN.UTF-8')
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S %A")

sender_email = "uyme@qq.com"
password = "1111111111"
receiver_email = "aooboo@88.com"

message = MIMEMultipart()
message["Subject"] = f"测试邮件{now}"
message['From'] = formataddr(('机器人-96', sender_email))
message['To'] = formataddr(('', receiver_email))

def add_attachment(message, file_name):
    with open(file_name, "rb") as attachment:
        att = MIMEApplication(attachment.read(), _subtype=file_name.split(".")[-1])
        att.add_header("Content-Disposition", "attachment", filename=file_name)
        message.attach(att)

def add_img(message, img_name, imageid):
    """添加图片；imageid 用于在正文中引用"""
    with open(img_name, "rb") as f:
        img = MIMEImage(f.read())
        message.attach(img)
        img.add_header('Content-ID', f'<{imageid}>')
    return f"<img src='cid:{imageid}'>"

add_attachment(message, "ccc.py")
image1 = add_img(message, "20230410113916.png", 'aadd')

body = f'<h1>这是一张图片</h1>：<br><br><br>{image1}'
message.attach(MIMEText(body, 'html'))

with smtplib.SMTP("smtp.qq.com", 587) as smtp:
    smtp.starttls()
    smtp.login(sender_email, password)
    smtp.sendmail(sender_email, receiver_email, message.as_string())

from wordcloud import WordCloud
from PIL import Image
import numpy as np

# 蒙版框架
mask = np.array(Image.open("123.png"))

wordcloud = WordCloud(background_color="white",
                      width=1800,
                      height=1600,
                      max_words=200,
                      max_font_size=100,
                      mask=mask,
                      contour_width=3,
                      contour_color='steelblue',
                      font_path="仓耳舒圆体W05.ttf")

# 文字内容
wordcloud.generate('''建立声望
社交与人脉
让自己的技能跟得上时代
''')

wordcloud.to_file('lz9.png')

import csv
import requests
from jira import JIRA

JIRA_SERVER = 'https://jira.mibo.com'
Personal_Access_Tokens = "填写令牌"

jira = JIRA(JIRA_SERVER, token_auth=Personal_Access_Tokens)
issues = jira.search_issues(
    jql_str="assignee was in (mibo) AND issuetype in (客户反馈, 缺陷)",
    json_result=True,
    fields='summary,assignee,status,created,resolutiondate,assignee,customfield_10313,reporter,issuetype',
    maxResults=500000)

def use_api():
    """直接使用 REST API 查询"""
    headers = {
        "Content-Type": "application/json",
        "authorization": f"Bearer {Personal_Access_Tokens}"
    }
    jira_url = ("https://jira.mibo.com/rest/shdsd-Timesheet/latest/workReport/extension2"
                "?queryType=convention&start=2022-01-01&end=2022-02-01&groupBy=U,P,I"
                "&userGroup=产品研发线/XXX中心/XXX部门")
    response = requests.get(jira_url, headers=headers).json()

def result_to_csv(issues, csv_file_path='issues.csv'):
    with open(csv_file_path, 'w', newline='', encoding="utf8") as file:
        writer = csv.writer(file)
        writer.writerow(['Key', 'Summary', 'Assignee', 'Status', "created", "fixtime",
                         "assignee", "fixer", "reporter", "type"])
        for issue in issues['issues']:
            try:
                fixer = issue['fields']['customfield_10313']['displayName']
            except Exception:
                fixer = "NONE"
            try:
                fixtime = issue['fields']['resolutiondate'][0:19]
            except Exception:
                fixtime = "NONE"
            created = issue['fields']['created'][0:19]
            assignee = issue['fields']['assignee']['displayName']
            reporter = issue['fields']['reporter']['displayName']
            type = issue['fields']['issuetype']['name']

            writer.writerow([issue['key'], issue['fields']['summary'],
                             issue['fields']['assignee']['displayName'],
                             issue['fields']['status']['name'], created, fixtime,
                             assignee, fixer, reporter, type])

import fitz
import sqlite3
import pywebio
from pywebio import config
from pywebio.input import file_upload, input_group
from pywebio.output import put_file
import json
import requests

class maxai:
    def __init__(self, auth="application-ad80edba84c30eb1161ae04e71ee7f78"):
        self.base_url = 'http://ai.tech.intra.com'
        self.auth = auth
        self.headers = {'accept': 'application/json', 'AUTHORIZATION': self.auth}
        self.new_chat = False
        self.chatid = self.create_chat()
        print(f"\r登录成功-{self.chatid}")

    def chat(self, query):
        chat_id = self.create_chat() if self.new_chat else self.chatid
        url = self.base_url + "/api/application/chat_message/" + chat_id
        data = {"message": query, "re_chat": False, "stream": True}
        response = requests.post(url, headers=self.headers, json=data, stream=True)
        translated_text = ''
        if response.status_code == 200:
            response.encoding = 'utf-8'
            for line in response.iter_lines(decode_unicode=True):
                if line:
                    event_data = json.loads(line[5:])
                    if event_data['is_end'] is False:
                        text = event_data['content']
                        if text:
                            translated_text += text
        return translated_text

    def create_chat(self):
        id = self.login_and_get_appinfo()
        url = self.base_url + "/api/application/" + id + "/chat/open"
        return requests.get(url, headers=self.headers).json()['data']

    def login_and_get_appinfo(self):
        url = self.base_url + '/api/application/profile'
        return requests.get(url, headers=self.headers).json()['data']['id']

def convert_color(color_int):
    """将颜色整数转换为 RGB 格式"""
    r = (color_int >> 16) & 0xFF
    g = (color_int >> 8) & 0xFF
    b = color_int & 0xFF
    return (r / 255, g / 255, b / 255)

def init_db():
    conn = sqlite3.connect('translations_cache.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS translations (
            original_text TEXT PRIMARY KEY,
            translated_text TEXT
        )
    ''')
    conn.commit()
    conn.close()

def get_cached_translation(text):
    conn = sqlite3.connect('translations_cache.db')
    cursor = conn.cursor()
    cursor.execute('SELECT translated_text FROM translations WHERE original_text = ?', (text,))
    result = cursor.fetchone()
    conn.close()
    return result[0] if result else None

def cache_translation(original_text, translated_text):
    conn = sqlite3.connect('translations_cache.db')
    cursor = conn.cursor()
    cursor.execute('INSERT INTO translations (original_text, translated_text) VALUES (?, ?)',
                   (original_text, translated_text))
    conn.commit()
    conn.close()

def translate_text(text):
    cached_translation = get_cached_translation(text)
    if cached_translation:
        return cached_translation
    return ccaam.chat(text)

init_db()
ccaam = maxai()

def trancpdf(pdf_document, filename):
    for page_num in range(pdf_document.page_count):
        page = pdf_document.load_page(page_num)
        blocks2 = page.get_text("dict", flags=11)["blocks"]
        for block in blocks2:
            for line in block["lines"]:
                for span in line["spans"]:
                    text = span["text"]
                    if text.strip() and len(text) > 10:
                        translated_text = translate_text(text)
                        if "抱歉" in translated_text or "翻译" in translated_text:
                            continue
                        font_size = span["size"]
                        color = span["color"]
                        bbox = span["bbox"]
                        rect = fitz.Rect(bbox)
                        page.insert_text((rect.x0, rect.y0), translated_text,
                                         fontsize=font_size / 3,
                                         color=convert_color(color),
                                         fontname="SimSun", fontfile="SimSun.ttf")
                        pdf_document.save("zh_CN/" + filename)
        content = open("zh_CN/" + filename, 'rb').read()
        put_file(filename, content, 'download me')

def sendmd():
    data = input_group("翻译pdf", [
        file_upload(accept=".pdf", placeholder="翻译pdf", name="file"),
    ])
    f = data['file']
    filename = f['filename']
    open(filename, 'wb').write(f['content'])
    if "file" in data:
        pdf_document = fitz.open(filename)
        trancpdf(pdf_document, filename)
        pdf_document.close()

if __name__ == '__main__':
    config(title="工具", theme="yeti", description="yes")
    pywebio.start_server(sendmd, port=1008, cdn=False)

import json
import urllib.parse
import sqlite3
from ai.deep_speek import maxai
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

apkey = 'application-24f27892e5f9da86d669d842c1c68254'
c1 = maxai(new_chat=True, auth=apkey)

def create_table(db_name):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS my_table (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        value TEXT NOT NULL
    )
    ''')
    conn.commit()
    cursor.close()
    conn.close()

def add_record(db_name, value):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    try:
        cursor.execute("INSERT INTO my_table (value) VALUES (?)", (value,))
        conn.commit()
    except sqlite3.Error as e:
        print(f"An error occurred while adding record: {e}")
    finally:
        cursor.close()
        conn.close()

def get_and_delete_min_id_value(db_name):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT id, value FROM my_table ORDER BY id ASC LIMIT 1")
        row = cursor.fetchone()
        if row:
            min_id, value = row
            cursor.execute("DELETE FROM my_table WHERE id = ?", (min_id,))
            conn.commit()
            return value
        return None
    except sqlite3.Error as e:
        print(f"An error occurred: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

class SSEHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

    def do_POST(self):
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length)
        decoded_str = urllib.parse.unquote(post_data.decode('utf-8'))
        add_record(db_name, decoded_str)

        self.send_response(200)
        self.send_header('Content-type', 'text/plain')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Cache-Control', 'no-cache')
        self.end_headers()
        self.wfile.write(b'')

    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-type", "text/html; charset=utf-8")
        self.send_header('Cache-Control', 'no-cache')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        value = get_and_delete_min_id_value(db_name)
        if value:
            response = c1.webchat(value)
            if response.status_code == 200:
                response.encoding = 'utf-8'
                for line in response.iter_lines(decode_unicode=True):
                    if line:
                        event_data = json.loads(line[5:])
                        if event_data['is_end'] is False:
                            text = event_data['content']
                            if text:
                                self.wfile.write(text.encode('utf-8'))

def run(server_class=ThreadingHTTPServer, handler_class=SSEHandler, port=8000):
    httpd = server_class(('', port), handler_class)
    print(f'Starting httpd server on port {port}...')
    httpd.serve_forever()

if __name__ == '__main__':
    db_name = 'ai'
    create_table(db_name)
    run()

import pyotp

def verify_2af():
    # 生成一个 2FA 密钥
    secret = pyotp.random_base32(64)
    # 生成基于密钥的验证码，有效时长 60 秒
    totp = pyotp.TOTP(secret, interval=60)

    code = totp.now()
    print(f"Code: {code}\nsecret: {secret}\n")
    user_input = input("输入验证码: ")
    if totp.verify(user_input):
        print("2AF Code is 有效.")
    else:
        print("2AF Code is 无效.")

if __name__ == "__main__":
    verify_2af()

import paramiko

ssh_client = paramiko.SSHClient()
# 自动把未知主机名与密钥加入本地 HostKeys，必须在 connect 之前调用
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname='192.168.137.105', port=22, username='root', password='123456')
stdin, stdout, stderr = ssh_client.exec_command('df -hT')
print(stdout.read().decode('utf-8'))
ssh_client.close()

# 密钥方式
private = paramiko.RSAKey.from_private_key_file('/root/.ssh/id_rsa')
ssh_client.connect(hostname='192.168.137.100', port=22, username='root', pkey=private)

tran = paramiko.Transport(('192.168.137.100', 22))
tran.connect(username="root", password='123456')
sftp = paramiko.SFTPClient.from_transport(tran)
sftp.put("/home/1.txt", "/tmp/1.txt")
sftp.get("/tmp/1.txt", "/home/1.txt")
tran.close()

import paramiko
import time

class UseSSH:
    def __init__(self):
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        private_key_path = "C:\\Users\\MIBO\\.ssh\\id_rsa"
        ssh.connect(hostname="10.44.3.66", username="root", key_filename=private_key_path)
        self.objssh = ssh

    def exec_cmd(self, cmd):
        stdin, stdout, stderr = self.objssh.exec_command(cmd)
        print(stdout.read().decode())

    def exec_invoke_cmd(self, k_andv: dict):
        channel = self.objssh.invoke_shell()
        for cmd, out in k_andv.items():
            channel.send(cmd + '\n')
            output = ""
            while not channel.recv_ready():   # 无可读数据时阻塞
                time.sleep(0.5)
            while channel.recv_ready():
                time.sleep(0.5)               # 有些机器反应慢，多等一会
                output += channel.recv(4096).decode()
            print(f"** {cmd} - {'执行成功' if out in output else '执行失败'}")
            print(output)
        channel.close()

    def __del__(self):
        self.objssh.close()

import socks, socket, paramiko

socks.set_default_proxy(socks.SOCKS5, "10.11.22.31", 65157)
socket.socket = socks.socksocket

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(hostname='192.168.124.191', port=22, username='root', password='anaxh2wn')
print(ssh.exec_command('ifconfig')[1].read().decode('utf-8'))
ssh.close()

import subprocess

completed = subprocess.run(['ls', '-1'])
print('returncode:', completed.returncode)

ret, val = subprocess.getstatusoutput("ping www.baidu.com")

sh = shelve.open("shelve")
sh["score"] = [80, 80, 80]
sh.close()

sh = shelve.open("shelve")
sh["score"].append(90)   # 不生效
sh.close()

combined = zip(x, y)
x, y = zip(*sorted(combined, key=lambda x: x[1], reverse=True))

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

def is_even(number):
    return number % 2 == 0

even_numbers = filter(is_even, numbers)
print(list(even_numbers))  # 输出 [2, 4, 6, 8, 10]

def mul(n):
    return n * n

num = [1, 2, 3, 4, 5]
result = list(map(mul, num))  # 使用 map 对列表 num 中每个元素应用 mul，返回新列表
print(result)  # [1, 4, 9, 16, 25]

import time

def auto_expiring_cache(expiration=300):
    cache = {}
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 清除过期缓存
            current_time = time.time()
            for key, value in list(cache.items()):  # list() 复制缓存条目，便于循环中删除
                if current_time - value['timestamp'] >= expiration:
                    del cache[key]
            # 生成缓存键
            key = (func.__name__, str(args), str(kwargs))
            print(key)
            # 检查缓存中是否已有该键
            if key in cache:
                return cache[key]['value']
            # 调用原始函数
            result = func(*args, **kwargs)
            # 更新缓存
            cache[key] = {
                'timestamp': current_time,
                'value': result
            }
            return result
        return wrapper
    return decorator

@auto_expiring_cache(300)
def ccc(x, y, z):
    print(f'计算中...{z}')
    return x + y

def iterate(n):
    i = 0
    while i < n:
        yield i
        i += 1

for i in iterate(10):
    print(i)

import traceback

def decorator(func):
    # 装饰器方法：函数执行失败时打印错误信息再抛出异常
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 打印出错的位置
            print(traceback.format_exc())
    return wrapper

import threading
import time

# 第一个工作函数
def worker1():
    for i in range(5):
        print("Worker 1: ", i)
        time.sleep(1)

# 第二个工作函数
def worker2():
    for i in range(15):
        print("Worker 2: ", i)

# 创建两个线程
t1 = threading.Thread(target=worker1)
t2 = threading.Thread(target=worker2)

# 启动线程
t1.start()
t2.start()

# 等待线程结束
t1.join()
t2.join()

f1 = open(file='read.txt', encoding='utf-8')
f1.read(5)
print(f1.tell())   # 指针位置为 5
f1.seek(9)         # 设置指针位置为 9
print(f1.tell())   # 指针位置为 9
f1.close()

import logging
from logging.config import fileConfig

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    filename='example.log', encoding='utf-8', level=logging.WARNING)
logging.basicConfig(format='%(asctime)s %(funcName)s%(lineno)d - %(levelname)s - %(message)s',
                    datefmt='%H:%M:%S')

conf = fileConfig("logging.conf")