Chemmy's Blog

import requests

# DeepSeek API 配置
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
DEEPSEEK_API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  # 替换为你的 API Key

def get_article_from_deepseek(my_describe):
    """
    调用 DeepSeek API 生成文章。
    :param my_describe: 给 AI 的提示词，描述文章要求。
    :return: 生成的文本内容，失败则返回 None。
    """
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }
    data = {
        "model": "deepseek-chat",  # 指定模型
        "messages": [
            {
                "role": "user",
                "content": my_describe
            }
        ]
    }
    try:
        response = requests.post(DEEPSEEK_API_URL, headers=headers, json=data)
        if response.status_code == 200:
            result = response.json()
            article = result["choices"][0]["message"]["content"]
            return article
        else:
            print(f"DeepSeek 请求失败，状态码：{response.status_code}，错误信息：{response.text}")
    except requests.RequestException as e:
        print(f"DeepSeek 请求发生错误：{e}")
    return None

# 示例：让 AI 基于近期热点创作
prompt = """
请查询最近三天内的热搜榜，取其中的一个热点，生成一篇公众号文章。
要求：
1. 标题要吸引人，有争议性。
2. 内容完整，结构清晰（有引言、主体、结论）。
3. 不能涉及侵权行为。
4. 输出纯文本，无需 Markdown 格式。
"""
article_content = get_article_from_deepseek(prompt)
if article_content:
    print("文章生成成功！")
    print(article_content[:500])  # 预览前500字符

# 微信公众号配置
WECHAT_APPID = "你的AppID"
WECHAT_APPSECRET = "你的AppSecret"
WECHAT_API_URL = "https://api.weixin.qq.com"

def get_wechat_access_token():
    """获取微信公众号的 access_token。"""
    url = f"{WECHAT_API_URL}/cgi-bin/token?grant_type=client_credential&appid={WECHAT_APPID}&secret={WECHAT_APPSECRET}"
    try:
        response = requests.get(url)
        if response.status_code == 200:
            result = response.json()
            if 'access_token' in result:
                return result['access_token']
            else:
                print(f"获取 access_token 失败，错误信息：{result}")
    except requests.RequestException as e:
        print(f"获取 access_token 发生错误：{e}")
    return None

def upload_cover_image(access_token, image_path):
    """上传封面图片并获取 media_id。"""
    url = f"{WECHAT_API_URL}/cgi-bin/media/upload?access_token={access_token}&type=image"
    with open(image_path, 'rb') as file:
        files = {'media': file}
        try:
            response = requests.post(url, files=files)
            if response.status_code == 200:
                result = response.json()
                if 'media_id' in result:
                    return result['media_id']
                else:
                    print(f"上传封面图片失败，错误信息：{result}")
        except requests.RequestException as e:
            print(f"上传封面图片发生错误：{e}")
    return None

def create_article_material(access_token, article, thumb_media_id):
    """将文章内容创建为公众号素材，并返回素材 media_id。"""
    url = f"{WECHAT_API_URL}/cgi-bin/media/uploadnews?access_token={access_token}"
    # 注意：公众号文章内容需要是 HTML 格式
    data = {
        "articles": [
            {
                "title": "AI生成的热点文章",  # 文章标题
                "thumb_media_id": thumb_media_id,  # 封面图片素材 ID
                "author": "AI Writer",  # 作者
                "digest": article[:100] + "...",  # 文章摘要（取前100字符）
                "show_cover_pic": 0,  # 是否在正文显示封面图片
                "content": article,
                "content_source_url": ""  # 原文链接（可留空）
            }
        ]
    }
    try:
        response = requests.post(url, json=data)
        if response.status_code == 200:
            result = response.json()
            if 'media_id' in result:
                return result['media_id']
            else:
                print(f"创建文章素材失败，错误信息：{result}")
    except requests.RequestException as e:
        print(f"创建文章素材发生错误：{e}")
    return None

def publish_article(access_token, media_id):
    """将已创建的素材发布给所有粉丝。"""
    url = f"{WECHAT_API_URL}/cgi-bin/message/mass/sendall?access_token={access_token}"
    data = {
        "filter": {
            "is_to_all": True  # 发送给所有用户
        },
        "mpnews": {
            "media_id": media_id
        },
        "msgtype": "mpnews"
    }
    try:
        response = requests.post(url, json=data)
        if response.status_code == 200:
            result = response.json()
            if result.get('errcode') == 0:
                print("文章发布成功！")
            else:
                print(f"文章发布失败，错误信息：{result}")
    except requests.RequestException as e:
        print(f"文章发布发生错误：{e}")

import os

def main():
    """主函数：一键生成并发布文章。"""
    print("开始自动创作与发布流程...")
    
    # 1. AI 生成文章
    prompt = "请以‘职场人的下班自由’为主题，写一篇800字左右的公众号文章，要求观点鲜明，引发共鸣。"
    print("正在调用 AI 生成文章...")
    article_content = get_article_from_deepseek(prompt)
    
    if not article_content:
        print("文章生成失败，流程终止。")
        return
    
    # 2. 处理编码与格式（防止乱码）
    # 公众号 API 要求内容为 UTF-8 编码的字符串
    article_content = article_content.encode('utf-8').decode('utf-8')
    # 可选：将纯文本转换为简单的 HTML 段落格式，以提升公众号阅读体验
    html_content = "<p>" + article_content.replace("\n", "</p><p>") + "</p>"
    
    # 3. 获取公众号访问令牌
    print("正在获取微信公众号访问令牌...")
    access_token = get_wechat_access_token()
    if not access_token:
        print("获取访问令牌失败，流程终止。")
        return
    
    # 4. 上传封面图片（需提前准备一张 cover.jpg）
    cover_image_path = "./cover.jpg"
    if not os.path.exists(cover_image_path):
        print(f"封面图片不存在：{cover_image_path}，请准备一张封面图。")
        # 此处可以改为使用默认图或跳过
        return
    print("正在上传封面图片...")
    thumb_media_id = upload_cover_image(access_token, cover_image_path)
    if not thumb_media_id:
        print("上传封面图片失败，流程终止。")
        return
    
    # 5. 创建文章素材
    print("正在创建公众号文章素材...")
    media_id = create_article_material(access_token, html_content, thumb_media_id)
    if not media_id:
        print("创建文章素材失败，流程终止。")
        return
    
    # 6. 发布文章
    print("正在发布文章...")
    publish_article(access_token, media_id)
    
    print("自动化流程执行完毕！")

if __name__ == "__main__":
    main()

# 读取视频流
strem_capture = cv2.VideoCapture("rtst://192.168.0.0/live/demo")

# 读取视频文件
file_capture = cv2.VideoCapture('demo.mp4')

# 读取摄像头
capture = cv2.VideoCapture(0)

# 校验获取VideoCapture类实例
if not capture.isOpened():
	return

# 获取视频帧的宽
width = capture.get(cv2.CAP_PROP_FRAME_WIDTH)

# 获取视频帧的高
height = cpature.get(cv2.CAP_PROP_FRAME_HEIGHT)

# 获取视频帧率
fps = capture.get(cv2.CAP_PROP_FPS)

success, frame = capture.read()

success_1 = capture.grab()
success_2 = stream_capture.grab()

if success_1 and success_2:
	frame_1 = capture.retrieve()
	frame_2 = stream_capture.retrieve()

# 设置摄像头分辨率的宽
capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)

# 设置摄像头分辨率的高
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)

writer = cv2.VideoWriter('output.mp4', 
	cv2.VideoWriter_fourcc(*'MP4V'), 30, (1080,1920))
writer.write(frame)

# 释放VideoCapture资源
capture.release()

# 释放VideoWriter资源
writer.release()

# -*- coding: utf-8 -*-
# /usr/bin/env/python3

import cv2
import time

capture = cv2.VideoCapture('rtsp://192.168.0.0/live/demo)
fourcc = cv2.VideoWriter_fourcc(*'MP$v') # 或H264,H265
fps = capture.get(cv2.CAP_PROP_FPS)
width = capture.get(cv2.CAP_PROP_FRAME_WIDTH)
height = capture.get(cv2.CAP_PROP_FRAME_HEIGHT)

writer = cv2.VideoWriter('demo.mp4', fourcd, fps, (height, width))


while True:
	if not capture.isOpened():
		time.sleep(0.5)
		continue
	success, frame = capture.read()
	if success:
		cv2.imshow('DEMO', frame) # 显示画面
		writer.write(frame) # 保存视频文件
	if (cv2.waitKey(20) & 0xff) == ord('q'): # 等待20ms并判断是否按下'q'退出,waitkey只能传入整数，
		break

capture.release() # 释放VideoCapture
writer.release() # 释放VideoWriter
cv2.destroyAllWindows() # 销毁所有opencv显示窗口