Xtao-Labs/cqwu-ehall
3
0
Fork 0
mirror of https://github.com/cqwu-ehall/cqwu-ehall.git synced 2024-11-22 02:55:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

✨ Support get score

Browse Source 
Co-authored-by: brian <brian@xtaolabs.com>
This commit is contained in:
brian 2023-03-08 20:48:12 +08:00 committed by GitHub
parent 499fd90264
commit 60c32861ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 262 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cqwu/client.py
View File

@ -1,7 +1,6 @@
import asyncio
from typing import Coroutine, Optional
from httpx import AsyncClient, Cookies
from urllib.parse import urlparse
from cqwu.methods import Methods
from cqwu.types import User
@ -28,19 +27,8 @@ class Client(Methods):
else:
self.host = "http://ehall.cqwu.edu.cn"
self.auth_host = "http://authserver.cqwu.edu.cn"
self.headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Host": urlparse(self.host).netloc,
"Origin": self.host,
"User-Agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.141 Safari/537.36"
}
self.cookies = Cookies()
self.sub_cookies = Cookies()
self.init_sub_web = []
self.request = AsyncClient
self.request = AsyncClient()
self.loop = asyncio.get_event_loop()
self.me: Optional[User] = None

1
cqwu/enums/__init__.py
View File

@ -1 +0,0 @@
from .order import OrderStatus

9
cqwu/enums/order.py
View File

@ -1,9 +0,0 @@
from enum import Enum
class OrderStatus(Enum):
NO_PAY = 0
NO_PAY_RE = 1
SUCCESS = 2
FAILURE = 3
EXPIRED = 4

17
cqwu/errors/auth.py
View File

@ -1,6 +1,19 @@
class UsernameOrPasswordError(Exception):
from .base import CQWUEhallError
class AuthError(CQWUEhallError):
pass
class CookieError(Exception):
class UsernameOrPasswordError(AuthError):
pass
class CookieError(AuthError):
pass
class NeedCaptchaError(AuthError):
""" 需要验证码才能登录 """
def __init__(self, captcha: bytes):
self.captcha = captcha

3
cqwu/errors/base.py Normal file
View File

@ -0,0 +1,3 @@
class CQWUEhallError(Exception):
"""Base class for exceptions in this module."""
pass

4
cqwu/methods/__init__.py
View File

@ -1,11 +1,13 @@
from .auth import Auth
from .epay import EPay
from .users import Users
from .xg import XG
class Methods(
Auth,
EPay,
Users
Users,
XG,
):
pass

12
cqwu/methods/auth/__init__.py
View File

@ -1,15 +1,19 @@
from .login_with_password import LoginWithPassword
from .check_captcha import CheckCaptcha
from .export_cookie_to_file import ExportCookieToFile
from .login import Login
from .login_with_cookie import LoginWithCookie
from .login_with_cookie_file import LoginWithCookieFile
from .export_cookie_to_file import ExportCookieToFile
from .login_with_password import LoginWithPassword
from .oauth import Oauth
class Auth(
LoginWithPassword,
CheckCaptcha,
ExportCookieToFile,
Login,
LoginWithCookie,
LoginWithCookieFile,
ExportCookieToFile,
LoginWithPassword,
Oauth
):
pass

34
cqwu/methods/auth/check_captcha.py Normal file
View File

@ -0,0 +1,34 @@
import time
import cqwu
from cqwu.errors.auth import NeedCaptchaError
class CheckCaptcha:
async def check_captcha(
self: "cqwu.Client",
username: int = None,
show_qrcode: bool = True,
):
""" 检查是否需要验证码 """
username = username or self.username
params = {
"username": username,
"pwdEncrypt2": "pwdEncryptSalt",
"_": str(round(time.time() * 1000))
}
url = f"{self.auth_host}/authserver/needCaptcha.html"
captcha_html = await self.request.get(url, params=params, follow_redirects=False)
if captcha_html.text == 'true':
params = {
"ts": str(round(time.time()))
}
captcha_url = f"{self.auth_host}/authserver/captcha.html"
res = await self.request.get(captcha_url, params=params, follow_redirects=False)
if not show_qrcode:
raise NeedCaptchaError(res.content)
with open("captcha.jpg", mode="wb") as f:
f.write(res.content)
print("验证码已保存在当前目录下的 captcha.jpg 文件中。")
return self.get_input("验证码")
return False

22
cqwu/methods/auth/login.py Normal file
View File

@ -0,0 +1,22 @@
import contextlib
from os.path import exists
import cqwu
from cqwu.errors.auth import CookieError
class Login:
async def login(
self: "cqwu.Client",
):
""" 登录 """
with contextlib.suppress(CookieError):
if self.cookie:
await self.login_with_cookie()
elif exists(self.cookie_file_path):
await self.login_with_cookie_file()
return
if self.username and self.password:
await self.login_with_password()
else:
raise CookieError()

12
cqwu/methods/auth/login_with_cookie.py
View File

@ -5,13 +5,15 @@ from cqwu.errors.auth import CookieError
class LoginWithCookie:
async def login_with_cookie(
self: "cqwu.Client",
cookie: str = None,
):
"""
使用 cookie 登录
"""
if not self.cookie:
cookie = cookie or self.cookie
if not cookie:
raise CookieError()
self.cookie = cookie # noqa
try:
data = self.cookie.split(";")
for cookie in data:
@ -19,7 +21,7 @@ class LoginWithCookie:
continue
key, value = cookie.split("=")
self.cookies.set(key, value)
self.sub_cookies.set(key, value)
self.request.cookies.set(key, value)
self.me = await self.get_me() # noqa
except:
raise CookieError()
except Exception as e:
raise CookieError() from e

4
cqwu/methods/auth/login_with_cookie_file.py
View File

@ -18,5 +18,5 @@ class LoginWithCookieFile:
with open(self.cookie_file_path, "r") as f:
self.cookie = f.read() # noqa
await self.login_with_cookie()
except:
raise CookieError()
except Exception as e:
raise CookieError() from e

59
cqwu/methods/auth/login_with_password.py
View File

@ -1,7 +1,3 @@
import time
from datetime import datetime
from urllib.parse import urlencode, urlparse
from lxml import etree
import cqwu
@ -12,24 +8,21 @@ from cqwu.utils.auth import encode_password
class LoginWithPassword:
async def login_with_password(
self: "cqwu.Client",
captcha_code: str = None,
show_qrcode: bool = True,
):
"""
使用学号加密码登录
"""
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Cache-Control': 'max-age=0',
'Connection': 'keep-alive',
'DNT': '1',
'Upgrade-Insecure-Requests': '1',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.5005.167 Safari/537.36',
"Referer": f"{self.auth_host}/authserver/login",
"Origin": self.auth_host,
"Host": urlparse(self.auth_host).netloc
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.63',
}
session = self.request(headers=self.headers, follow_redirects=True)
html = await session.get(f"{self.host}/authserver/login")
html = await self.request.get(f"{self.auth_host}/authserver/login", headers=headers, follow_redirects=True)
self.cookies.update(html.cookies)
tree = etree.HTML(html.text)
pwd_default_encrypt_salt = tree.xpath('//*[@id="pwdDefaultEncryptSalt"]/@value')[0]
@ -42,33 +35,27 @@ class LoginWithPassword:
'_eventId': tree.xpath('//*[@id="casLoginForm"]/input[4]/@value')[0],
'rmShown': tree.xpath('//*[@id="casLoginForm"]/input[5]/@value')[0]
}
# 是否需要验证码
params = {
"username": self.username,
"pwdEncrypt2": "pwdEncryptSalt",
"_": str(round(time.time() * 1000))
}
need_captcha_url = f"{self.auth_host}/authserver/needCaptcha.html?{urlencode(params)}"
async with self.request() as client:
html = await client.get(need_captcha_url, follow_redirects=False)
if html.text == 'true':
ts = round(datetime.now().microsecond / 1000) # get milliseconds
captcha_url = f"{self.auth_host}/authserver/captcha.html?" + urlencode({"ts": ts})
async with self.request() as client:
res = await client.get(captcha_url, follow_redirects=False)
with open("captcha.jpg", mode="wb") as f:
f.write(res.content)
print("验证码已保存在当前目录下的 captcha.jpg 文件中。")
code = self.get_input("验证码")
form_data['captchaResponse'] = code
if not captcha_code:
form_data['captchaResponse'] = await self.check_captcha(show_qrcode=show_qrcode)
# 登录
async with self.request(headers=headers, cookies=self.cookies) as client:
html = await client.post(f"{self.auth_host}/authserver/login", data=form_data, follow_redirects=False)
headers = {
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Cache-Control': 'max-age=0',
'Connection': 'keep-alive',
'Origin': self.auth_host,
'Referer': f'{self.auth_host}/authserver/login',
'Upgrade-Insecure-Requests': '1',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.63',
}
html = await self.request.post(
f"{self.auth_host}/authserver/login",
headers=headers,
data=form_data,
follow_redirects=False,
)
if 'CASTGC' not in html.cookies.keys():
raise UsernameOrPasswordError
self.cookies.update(html.cookies)
self.sub_cookies.update(html.cookies)
self.me = await self.get_me() # noqa

7
cqwu/methods/auth/oauth.py
View File

@ -13,10 +13,5 @@ class Oauth:
使用 统一身份认证平台 登录子系统，并且保存 cookie
"""
host = host or urlparse(url).hostname
async with self.request(cookies=self.sub_cookies, follow_redirects=True) as client:
html = await client.get(url)
for history in html.history:
self.sub_cookies.update(history.cookies)
if host not in self.init_sub_web:
self.init_sub_web.append(host)
html = await self.request.get(url, follow_redirects=True)
return None if html.url.host != host else html

2
cqwu/methods/epay/__init__.py
View File

@ -1,11 +1,9 @@
from .gen_pay_qrcode import GenPayQrcode
from .get_balance import GetBalance
from .get_orders import GetOrders
class EPay(
GenPayQrcode,
GetBalance,
GetOrders
):
pass

8
cqwu/methods/epay/gen_pay_qrcode.py
View File

@ -2,6 +2,7 @@ import qrcode
from bs4 import BeautifulSoup
import cqwu
from cqwu.errors.auth import CookieError
class GenPayQrcode:
@ -11,9 +12,12 @@ class GenPayQrcode:
"""
生成支付二维码
"""
html = await self.oauth("http://218.194.176.214:8382/epay/thirdconsume/qrcode")
url = "http://218.194.176.214:8382/epay/thirdconsume/qrcode"
html = await self.oauth(url)
if not html:
return
raise CookieError()
if html.url != url:
raise CookieError()
soup = BeautifulSoup(html.text, "lxml")
try:
data = soup.find("input", attrs={"id": "myText"})["value"]

8
cqwu/methods/epay/get_balance.py
View File

@ -2,6 +2,7 @@ from typing import Optional
from bs4 import BeautifulSoup
import cqwu
from cqwu.errors.auth import CookieError
class GetBalance:
@ -14,9 +15,12 @@ class GetBalance:
Returns:
str: 余额
"""
html = await self.oauth("http://218.194.176.214:8382/epay/thirdapp/balance")
url = "http://218.194.176.214:8382/epay/thirdapp/balance"
html = await self.oauth(url)
if not html:
return ""
raise CookieError()
if html.url != url:
raise CookieError()
soup = BeautifulSoup(html.text, "lxml")
try:
return soup.find_all("div", "weui-cell__ft")[2].next

39
cqwu/methods/epay/get_orders.py
View File

@ -1,39 +0,0 @@
from typing import List
import cqwu
from cqwu.types.order import Order
class GetOrders:
async def get_orders(
self: "cqwu.Client",
page: int = 1,
page_size: int = 10,
search: str = ""
) -> List[Order]:
"""
获取历史订单
:param page: 页码
:param page_size: 每页数量
:param search: 搜索关键字
:return: 订单列表
"""
url = "http://pay.cqwu.edu.cn/queryOrderList"
params = {
"orderno": search,
"page": page,
"pagesize": page_size,
}
if "pay.cqwu.edu.cn" not in self.init_sub_web:
await self.oauth(
"http://authserver.cqwu.edu.cn/authserver/login?service="
"http%3A%2F%2Fpay.cqwu.edu.cn%2FsignAuthentication%3Furl%3DopenPortal")
async with self.request(cookies=self.sub_cookies, follow_redirects=True) as client:
html = await client.get(url, params=params)
try:
data = html.json()["payOrderList"]
except KeyError:
return []
return [Order(**order) for order in data]

11
cqwu/methods/users/get_me.py
View File

@ -3,6 +3,7 @@ from bs4 import BeautifulSoup
import cqwu
from cqwu import types
from cqwu.errors.auth import CookieError
def get_value_from_soup(soup: BeautifulSoup, attr_id: str) -> Union[type(None), str, int]:
@ -26,10 +27,14 @@ class GetMe:
Returns:
types.User: 个人信息
"""
html = await self.oauth(
"http://218.194.176.8/prizepunishnv/studentInfoManageStudentNV!forwardStudentInfo.action")
url = "http://218.194.176.8/prizepunishnv/studentInfoManageStudentNV!forwardStudentInfo.action"
html = await self.oauth(url)
if not html:
return None
raise CookieError()
if html.url != url:
html = await self.request.get(url)
if html.url != url:
raise CookieError()
soup = BeautifulSoup(html.text, "lxml")
data = {
"username": "detail_xh",

7
cqwu/methods/xg/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .get_score import GetScore
class XG(
GetScore,
):
pass

65
cqwu/methods/xg/get_score.py Normal file
View File

@ -0,0 +1,65 @@
import json
from typing import List
import cqwu
from cqwu import types
from cqwu.errors.auth import CookieError
class GetScore:
async def get_score(
self: "cqwu.Client",
year: int = 2022,
semester: int = 1,
) -> List["types.Score"]:
"""
获取期末成绩
Returns:
List[types.Score]: 成绩列表
"""
url = "http://xg.cqwu.edu.cn/xsfw/sys/zhcptybbapp/*default/index.do#/cjcx"
html = await self.oauth(url)
if not html:
raise CookieError()
if html.url != url:
raise CookieError()
await self.request.get(
"http://xg.cqwu.edu.cn/xsfw/sys/swpubapp/indexmenu/getAppConfig.do?appId=5275772372599202&appName=zhcptybbapp&v=046351851777942055")
query_url = "http://xg.cqwu.edu.cn/xsfw/sys/zhcptybbapp/modules/cjcx/cjcxbgdz.do"
headers = {
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'Origin': 'http://xg.cqwu.edu.cn',
'Referer': 'http://xg.cqwu.edu.cn/xsfw/sys/zhcptybbapp/*default/index.do',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36 Edg/110.0.1587.63',
'X-Requested-With': 'XMLHttpRequest',
}
query_data = [
{
"name": "XN",
"caption": "学年",
"linkOpt": "AND",
"builderList": "cbl_m_List",
"builder": "m_value_equal",
"value": str(year),
},
{
"name": "XQ",
"caption": "学期",
"linkOpt": "AND",
"builderList": "cbl_m_List",
"builder": "m_value_equal",
"value": str(semester),
}
]
data = {
'querySetting': json.dumps(query_data),
'pageSize': '100',
'pageNumber': '1',
}
html = await self.request.post(query_url, headers=headers, data=data)
data = [types.Score(**i) for i in html.json()["datas"]["cjcxbgdz"]["rows"]]
return data

1
cqwu/types/__init__.py
View File

@ -1 +1,2 @@
from .score import Score
from .user import User

27
cqwu/types/order.py
View File

@ -1,27 +0,0 @@
import cqwu
class Order:
def __init__(
self,
orderno: str = None,
payproname: str = None,
orderamt: float = None,
createtime: str = None,
payflag: str = "",
**_
):
"""
订单
:param orderno: 订单编号
:param payproname: 缴费项目
:param orderamt: 缴费金额
:param createtime: 订单生成时间
:param payflag: 缴费状态
"""
self.order_id = orderno
self.project = payproname
self.amount = orderamt / 100
self.create_time = createtime
self.status = cqwu.enums.OrderStatus(int(payflag))

43
cqwu/types/score.py Normal file
View File

@ -0,0 +1,43 @@
from pydantic import BaseModel
class Score(BaseModel):
""" 成绩类 """
KCMC: str
XF: float
ZCJ: float
JD: str
XN: str
XN_DISPLAY: str
XQ: str
XQ_DISPLAY: str
@property
def name(self) -> str:
""" 课程名称 """
return self.KCMC
@property
def credit(self) -> float:
""" 学分 """
return self.XF
@property
def score(self) -> float:
""" 成绩 """
return self.ZCJ
@property
def grade_point(self) -> float:
""" 绩点 """
return float(self.JD)
@property
def year(self) -> int:
""" 学年 """
return int(self.XN)
@property
def semester(self) -> int:
""" 学期 """
return int(self.XQ)

9
requirements.txt
View File

@ -1,6 +1,7 @@
httpx
httpx==0.23.3
lxml==4.9.1
PyExecJS==1.5.1
beautifulsoup4==4.11.1
qrcode==7.3.1
PyExecJS2==1.6.1
beautifulsoup4==4.11.2
qrcode==7.4.2
pillow
pydantic==1.10.5