diff --git a/weibo/main.py b/weibo/main.py new file mode 100644 index 0000000..e90b6e9 --- /dev/null +++ b/weibo/main.py @@ -0,0 +1,314 @@ +import hashlib +import math +import time +import zlib + +import requests +import re +import json + + +class Wb: + def __init__(self, cookie): + self.chunk_size = 1024 * 1024 * 8 + self.xsrf_token = None + if "XSRF-TOKEN" in cookie: + del cookie["XSRF-TOKEN"] + if "xsrf-token" in cookie: + del cookie["xsrf-token"] + self.cookie = cookie + self.headers = self.get_headers() + self.screen_name, self.idstr, self.xsrf_token = self.get_tk_info() + self.headers['x-xsrf-token'] = self.xsrf_token + self.cookie["XSRF-TOKEN"] = self.xsrf_token + + # self.headers[''] + + def md5(self, data): + if type(data) == str: + data = data.encode("utf-8") + return hashlib.md5(data).hexdigest() + + def get_cs(self, data): + return zlib.crc32(data) & 0xffffffff + + def get_session_id(self, video_data, vide_path): + data = f'{str(int(time.time() * 1000))}|{len(video_data)}|{vide_path}|video|{str(int(time.time()))}' + return hashlib.md5(data.encode()).hexdigest() + + def get_headers(self): + headers = { + "cache-control": "no-cache", + "pragma": "no-cache", + "priority": "u=0, i", + "sec-ch-ua": "\"Chromium\";v=\"9\", \"Not?A_Brand\";v=\"8\"", + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": "\"Windows\"", + "sec-fetch-dest": "document", + "sec-fetch-mode": "navigate", + "sec-fetch-site": "same-origin", + "sec-fetch-user": "?1", + "upgrade-insecure-requests": "1", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 SLBrowser/9.0.6.8151 SLBChan/10 SLBVPV/64-bit" + } + if self.xsrf_token: + headers['x-xsrf-token'] = self.xsrf_token + return headers + + def get_tk_info(self): + url = "https://weibo.com/" + response = requests.get(url, headers=self.headers, cookies=self.cookie) + data = json.loads(re.search('window\.\$CONFIG = (.*?);}', response.text).group(1)) + user = data['user'] + screen_name = user['screen_name'] + idstr = user['idstr'] + xsrf_token = response.cookies.get('XSRF-TOKEN') + print(screen_name, idstr, xsrf_token) + return screen_name, idstr, xsrf_token + + def put_image(self, image_data): + url = 'https://picupload.weibo.com/interface/upload.php?' + params = { + "file_source": "1", + "cs": self.get_cs(image_data), + "ent": "miniblog", + "appid": "339644097", + "uid": self.idstr, + "raw_md5": self.md5(image_data), + "ori": "1", + "mpos": "1", + "nick": self.screen_name, + "pri": "0", + "request_id": str(int(time.time() * 1000)), + "file_size": len(image_data) + } + res = requests.post(url=url, data=image_data, headers=self.headers, cookies=self.cookie, params=params).json() + print(res) + return res + + def get_media_group_id(self): + url = "https://weibo.com/ajax/multimedia/mediaGroupInit" + headers = self.get_headers() + headers.update( + { + "referer": "https://weibo.com/", + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "server-version": "v2026.03.04.2", + } + ) + response = requests.get(url, headers=headers, cookies=self.cookie).json() + print(response) + return response['data']['media_group_id'] + + def ini_video_id(self, media_group_id, video_name, video_data): + url = "https://fileplatform-cn1.api.weibo.com/2/fileplatform/init.json" + + params = { + "source": "339644097", + "size": len(video_data), + "name": video_name, + "type": "video", + "client": "web", + "session_id": self.get_session_id(video_data, video_name) + } + + boundary = f"2067456weiboPro{str(int(time.time()))}" + + headers = self.get_headers() + headers.update({ + + "Content-Type": f"multipart/mixed; boundary={boundary}", + + }) + + # 完全按 curl 构造原始 multipart body + body = ( + f"--{boundary}\r\n" + 'Content-Disposition: form-data; name="biz_file"\r\n\r\n' + '{"mediaprops":"{\\"screenshot\\":1,\\"media_group_id\\":\\"' + media_group_id + '\\"}"}\r\n' + f"--{boundary}--\r\n" + ) + + response = requests.post( + url, + params=params, + headers=headers, + cookies=cookies, + data=body.encode("utf-8") + ) + print(response.json()) + return response.json() + + def upload_video(self, upload_id, media_id, video_data, index, chunk_count, start_loc): + url = "https://up-cn1.video.weibocdn.com/2/fileplatform/upload.json?" + params = { + "source": "339644097", + "upload_id": upload_id, + "media_id": media_id, + "upload_protocol": "binary", + "type": "video", + "client": "web", + "check": self.md5(video_data), + "index": index, + "size": len(video_data), + "start_loc": start_loc, + "count": chunk_count + } + # print(params) + res = requests.post(url=url, params=params, headers=self.headers, cookies=cookies, data=video_data + ).json() + print(res) + return res + + def chak_video(self, upload_id, media_id, size, count): + url = "https://fileplatform-cn1.api.weibo.com/2/fileplatform/check.json" + data = { + "source": "339644097", + "upload_id": upload_id, + "media_id": media_id, + "upload_protocol": "binary", + "count": count, + "action": "finish", + "size": size, + "client": "web", + "status": "" + } + response = requests.post(url, headers=self.headers, cookies=cookies, data=data).json() + print(response) + return response + + def get_video_pid(self, media_id): + url = "https://weibo.com/ajax/multimedia/output" + params = { + "source": "339644097", + "ids": media_id, + "labels": "screenshot" + } + headers = self.get_headers() + headers.update({ + + "referer": "https://weibo.com/", + + }) + for i in range(10): + response = requests.get(url, headers=headers, cookies=cookies, params=params).json() + print(response) + if response['data'] == {}: + time.sleep(5) + continue + else: + return json.loads(response['data'][str(media_id)]['screenshot']['file_detail'])['files'][0]['file_id'] + return False + + def release(self, pic_id, content): + url = "https://weibo.com/ajax/statuses/update" + data = { + "content": content, + "visible": "0", + "share_id": "", + "vote": "", + "media": "", + "need_transcode": "1", + "pic_id": json.dumps(pic_id, ensure_ascii=False) + } + + headers = self.get_headers() + headers.update({ + + "origin": "https://weibo.com", + "pragma": "no-cache", + "priority": "u=1, i", + "referer": "https://weibo.com/", + "sec-ch-ua": "\"Chromium\";v=\"9\", \"Not?A_Brand\";v=\"8\"", + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": "\"Windows\"", + "sec-fetch-dest": "empty", + "sec-fetch-mode": "cors", + "sec-fetch-site": "same-origin", + "server-version": "v2026.03.04.2", + + } + ) + + response = requests.post(url, headers=headers, cookies=self.cookie, data=data).text + print(response) + return response + + def run_video(self, video_name): + media_group_id = self.get_media_group_id() + with open(video_name, "rb") as f: + video_data = f.read() + video_info = self.ini_video_id(media_group_id=media_group_id, video_name=video_name, video_data=video_data) + upload_id = video_info["upload_id"] + media_id = video_info["media_id"] + chunk_count = math.ceil(len(video_data) / self.chunk_size) + for index in range(chunk_count): + # start = index * self.chunk_size + # end = min((index + 1) * self.chunk_size, len(video_data)) + # f.seek(start) + # chunk_data = f.read(end - start) + # print(start, end) + # self.upload_video(upload_id=upload_id, media_id=media_id, video_data=chunk_data, index=index, + # chunk_count=chunk_count, start_loc=start) + start = index * self.chunk_size + end = min(start + self.chunk_size, len(video_data)) + f.seek(start) + chunk_data = f.read(end - start) + + self.upload_video( + upload_id=upload_id, + media_id=media_id, + video_data=chunk_data, + index=index, + chunk_count=chunk_count, + start_loc=start + ) + self.chak_video(upload_id=upload_id, media_id=media_id, size=len(video_data), count=chunk_count) + res = { + "type": "video", + "media_id": media_id, + "media_group_id": media_group_id + } + + pid = self.get_video_pid(media_id) + if pid: + res['pid'] = pid + return res + + def run_images(self, image_path): + with open(image_path, 'rb') as f: + pid = self.put_image(image_data=f.read())['pic']['pid'] + return { + "type": "image/png", + "pid": pid + } + + def run(self): + pid_list = [] + pid_list.append(self.run_images(r'D:\2026-3\image\2.png')) + pid_list.append(self.run_video(r'1.mp4')) + print(pid_list) + self.release(pic_id=pid_list, content='你们好我是2222') + + +if __name__ == '__main__': + cookies = { + "SCF": "AnPzE03xm8BB9HlgIFL6amrpHcKurEbxl_Lidx2mwHySmzyyoCMjVNxfKQCNBwYxr-gvT8-50DlpsHI_XJIJlqA.", + "SUB": "_2A25EosHEDeRhGeFG7lYX9SnKzT2IHXVn3lsMrDV8PUNbmtAbLRnGkW1NeV0u11k3Ss6dRbldGFr1JUHhVsbep0_3", + "SUBP": "0033WrSXqPxfM725Ws9jqgMF55529P9D9W5mUlS3fLogK2SEkeFqLNEa5NHD95QN1h-XSo-NSoqpWs4DqcjMi--NiK.Xi-2Ri--ciKnRi-zNS0nfShqfS0qceBtt", + "ALF": "02_1775124116", + "SINAGLOBAL": "8767812016817.392.1772592155581", + "ULV": "1772610236625:2:2:2:6590739668820.657.1772610236578:1772592155651", + "XSRF-TOKEN": "DQZxABOM_NfYDlXz1lTy1cOD", + "WBPSESS": "uH_pEmgfiICupYDn2MNJGBnQEfhqzo2bMY0VUuWK05-TR78aeKUmP0AyDqfzYmdbsI2dVYBedhIHV1snwDUpD1Lz1KXhMSCgjztzOlrwCTDLIhXWNZMTlGNIkU-lI5cmUf2721Pj2oQsCiRB-HQg1w==" + } + # Wb(cookies).get_video_pid('5272880565649510') + # print(Wb(cookies).run_video(video_name='1.mp4')) + # print(Wb(cookies).run_images(r'D:\2026-3\image\2.png')) + print(Wb(cookies).run()) + # Wb(cookies).release([{'type': 'image/png', 'pid': '008zzlgNgy1iavb8voqu5j30cu03iaa7'}, + # {'type': 'video', 'media_id': '5272882008227937', + # 'media_group_id': '0c22cd44-17db-11f1-b886-09e635c8dab8'}], + # '你们好')