ddrwode/lm_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rgfewfger

Browse Source
This commit is contained in:
27942
2025-11-19 15:59:38 +08:00
parent 198ef158ce
commit 6000303f30
5 changed files with 57 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
models/__init__.py
View File

@@ -1,7 +1,7 @@
from peewee import *
# 连接到 SQLite 数据库，如果文件不存在会自动创建
db = SqliteDatabase(r'E:\新建文件夹\lm_job\models\database.db')
# from peewee import *
#
# # 连接到 SQLite 数据库，如果文件不存在会自动创建
# db = SqliteDatabase(r'E:\新建文件夹\lm_job\models\database.db')
import pymysql
@@ -29,29 +29,29 @@ db1 = MySQLDatabase(
)
class BaseModel(Model):
class Meta:
database = db1
def save(self, *args, **kwargs):
"""在调用 save 时自动连接和关闭（若无事务）"""
db.connect(reuse_if_open=True)
try:
result = super().save(*args, **kwargs)
finally:
# 若当前没有事务且连接仍然打开，则关闭连接
if not db.in_transaction() and not db.is_closed():
db.close()
return result
@classmethod
def get_or_create(cls, defaults=None, **kwargs):
"""在调用 get_or_create 时自动连接和关闭（若无事务）"""
db.connect(reuse_if_open=True)
try:
obj, created = super().get_or_create(defaults=defaults, **kwargs)
finally:
# 若当前没有事务且连接仍然打开，则关闭连接
if not db.in_transaction() and not db.is_closed():
db.close()
return obj, created
# class BaseModel(Model):
# class Meta:
# database = db1
#
# def save(self, *args, **kwargs):
# """在调用 save 时自动连接和关闭（若无事务）"""
# db.connect(reuse_if_open=True)
# try:
# result = super().save(*args, **kwargs)
# finally:
# # 若当前没有事务且连接仍然打开，则关闭连接
# if not db.in_transaction() and not db.is_closed():
# db.close()
# return result
#
# @classmethod
# def get_or_create(cls, defaults=None, **kwargs):
# """在调用 get_or_create 时自动连接和关闭（若无事务）"""
# db.connect(reuse_if_open=True)
# try:
# obj, created = super().get_or_create(defaults=defaults, **kwargs)
# finally:
# # 若当前没有事务且连接仍然打开，则关闭连接
# if not db.in_transaction() and not db.is_closed():
# db.close()
# return obj, created

4
models/ips.py
View File

@@ -1,9 +1,9 @@
from peewee import *
from models import BaseModel, db1
from models import db1
class Ips(BaseModel):
class Ips(Model):
id = IntegerField(primary_key=True)
host = CharField(null=True)
port = CharField(null=True)

4
models/xstart.py
View File

@@ -1,9 +1,9 @@
from peewee import *
from models import db1, BaseModel
from models import db1
class Xstart(BaseModel):
class Xstart(Model):
id = AutoField(primary_key=True) # 自增主键
bit_id = IntegerField(null=True)
start = CharField(null=True)

6
models/xtoken.py
View File

@@ -1,9 +1,9 @@
from peewee import *
# 假设 db 已经在其他地方定义并连接到数据库
from models import BaseModel, db1
from models import db1
class XToken(BaseModel):
class XToken(Model):
id = AutoField(primary_key=True) # 自增主键
hub_id = IntegerField(null=True) # hub_id 字段，整型，可为空
start = IntegerField(null=True) # start 字段，整型，可为空
@@ -19,5 +19,3 @@ class XToken(BaseModel):
database = db1 # 所属数据库
table_name = 'x_token' # 表名
# if __name__ == '__main__':
# XToken.create_table()

45
推特/main.py
View File

@@ -1,12 +1,13 @@
import random
import time
import pyotp
import random
from loguru import logger
from concurrent.futures import ThreadPoolExecutor
from DrissionPage import ChromiumOptions, ChromiumPage
from peewee import fn
from bit_tools import openBrowser, createBrowser
from bit_tools import openBrowser, createBrowser, get_group_lists
from models.ips import Ips
from models.xstart import Xstart
from models.xtoken import XToken
@@ -32,8 +33,6 @@ class Hub_Web:
timeout=10,
)
# //*[@id="react-root"]/div/div/div[2]/header/div/div/div/div[2]/div/button/div[2]/div/div/div/div[1]/span/span
if user_name and self.x_info.user_name.lower() == user_name.text.split("@")[-1].lower():
return True
else:
@@ -61,14 +60,9 @@ class Hub_Web:
self.x_tab.ele('x://input[@name="password"]').input(self.x_info.password)
self.x_tab.ele('x://button[@data-testid="LoginForm_Login_Button"]').click()
self.two_tab = self.page.new_tab(url="https://2fa.fb.rip/")
self.two_tab.ele('x://input[@class="secret-block__input"]').input(self.x_info.two_fa)
# self.two_tab.ele('x://button[text()="Click for authentication code"]').click()
code = self.two_tab.ele('x://div[@id="verifyCode"]').text
self.x_tab.ele('x://input[@data-testid="ocfEnterTextTextInput"]').input(code)
# 创建一个 TOTP 对象
totp = pyotp.TOTP(self.x_info.two_fa)
self.x_tab.ele('x://input[@data-testid="ocfEnterTextTextInput"]').input(totp.now())
self.x_tab.ele('x://button[@data-testid="ocfEnterTextNextButton"]').click()
time.sleep(5)
@@ -76,8 +70,8 @@ class Hub_Web:
self.x_tab.ele('x://button[@data-testid="OAuth_Consent_Button"]').click()
if self.x_testing():
self.hub_web_info.x_start = 1
self.hub_web_info.save()
self.xstart_info.start = 1
self.xstart_info.save()
return True
else:
@@ -286,7 +280,10 @@ class Hub_Web:
logger.info("没有浏览器，创建中。。。")
fz_datas = get_group_lists()
bit_id = createBrowser(
groupId=fz_datas['推特'],
host=self.ips_info.host,
port=int(self.ips_info.port),
proxyUserName=self.ips_info.username,
@@ -302,7 +299,7 @@ class Hub_Web:
logger.error(f"推特名字：{self.x_info.user_name}，浏览器打开失败")
return
time.sleep(5)
# time.sleep(5)
for _, i in enumerate(self.page.get_tabs()):
if _ == 0:
@@ -311,22 +308,22 @@ class Hub_Web:
i.close()
if self.login_x_main():
self.hub_web_info.x_start = 1
self.hub_web_info.save()
self.xstart_info.start = 1
self.xstart_info.save()
logger.info(f"推特名字：{self.x_info.user_name}，登录x成功！！！")
logger.success(f"推特名字：{self.x_info.user_name}，登录x成功！！！")
time.sleep(25)
for i in self.x_tab.cookies():
if i["name"] == "auth_token":
self.hub_web_info.cookie = i["value"]
self.hub_web_info.save()
self.xstart_info.cookie = i["value"]
self.xstart_info.save()
else:
logger.error(f"推特名字：{self.x_info.user_name}，登录x失败！！！")
self.hub_web_info.x_start = 0
self.hub_web_info.save()
self.xstart_info.start = 0
self.xstart_info.save()
# try:
# self.soon_action()
@@ -342,13 +339,13 @@ class Hub_Web:
if __name__ == '__main__':
# 同时运行
max_threads = 1
max_threads = 10
delay_between_start = 10 # 每次启动线程之间的延迟时间（秒）
with ThreadPoolExecutor(max_workers=max_threads) as executor:
for x_token_info in XToken.select():
xstart_info = Xstart.get_or_create(
xstart_info, start = Xstart.get_or_create(
x_id=x_token_info.id,
)