Optimize Structure

main.py

@@ -3,61 +3,35 @@
 import re
 import time
 import traceback
-from base64 import b64encode
 
 import requests
-from Crypto.Cipher import PKCS1_v1_5 as Cipher_pksc1_v1_5
-from Crypto.PublicKey import RSA
-from ddddocr import DdddOcr
-
-from cap_denoise import dn
 
+from utility import encrypt, cap_recognize
 
 username = os.environ["USERNAME"]
 password = os.environ["PASSWORD"]
 
 if not (username and password):
     raise Exception("请设置Secret: USERNAME和PASSWORD")
 
-
-org_id = '172442'  # "北京市海淀团区委"
-# or check string type
 try:
-    org_id_input = os.environ["ORGID"]
-    if org_id_input:
-        org_id = int(org_id_input)
+    org_id = str(int(os.environ["ORGID"]))  # check type
 except:
-    ...
-
-
-def encrypt(t):
-    public_key = "-----BEGIN PUBLIC KEY-----\nMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQD5uIDebA2qU746e/NVPiQSBA0Q3J8/G23zfrwMz4qoip1vuKaVZykuMtsAkCJFZhEcmuaOVl8nAor7cz/KZe8ZCNInbXp2kUQNjJiOPwEhkGiVvxvU5V5vCK4mzGZhhawF5cI/pw2GJDSKbXK05YHXVtOAmg17zB1iJf+ie28TbwIDAQAB\n-----END PUBLIC KEY-----"
-    rsa_key = RSA.importKey(public_key)
-    cipher = Cipher_pksc1_v1_5.new(rsa_key)
-    cipher_text = b64encode(cipher.encrypt(t.encode()))
-    return cipher_text.decode()
-
-
+    org_id = '172442'
 
 url = ''
+ua = os.getenv('UA',
+               'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36 Edg/80.0.361.111')
 for _ in range(10):
     try:
         bjySession = requests.session()
-        # set session timeout
-        bjySession.timeout = 5
-        fake_ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36 Edg/80.0.361.111'
-        bjySession.headers.update({
-            "User-Agent": fake_ua, })
-
-        r = bjySession.get(url="https://m.bjyouth.net/site/login")
+        bjySession.timeout = 5  # set session timeout
+        bjySession.headers.update({"User-Agent": ua, })
+        touch = bjySession.get(url="https://m.bjyouth.net/site/login")
         cap_url = "https://m.bjyouth.net" + re.findall(
-            r'src="/site/captcha.+" alt=', r.text)[0][5:-6]
-
-        cap = bjySession.get(url=cap_url)
-        ocr = DdddOcr()
-        cap = dn(cap.content)
-        cap_text = ocr.classification(cap)
-        print(f'Captcha OCR: {cap_text}')
+            r'src="/site/captcha.+" alt=', touch.text)[0][5:-6]
+        cap_text = cap_recognize(bjySession.get(url=cap_url).content)
+        print(f'验证码识别: {cap_text}')
         _csrf_mobile = bjySession.cookies.get_dict()['_csrf_mobile']
         login_password = encrypt(password)
         login_username = encrypt(username)
@@ -69,25 +43,26 @@ def encrypt(t):
                                       'Login[verifyCode]': cap_text
                                   })
         if login_r.text == '8':
-            print('Login: 验证码错误')
-        print(f'Login: [{login_r.status_code}]{login_r.text}')
+            raise Exception('Login:识别的验证码错误')
+        print(f'Login:[{login_r.status_code}]{login_r.text}')
         r = json.loads(bjySession.get("https://m.bjyouth.net/dxx/index").text)
         if 'newCourse' not in r:
             print(r)
         url = r['newCourse']['url']
         title = r['newCourse']['title']
+        course_id = r['newCourse']['id']
         break
     except:
         time.sleep(3)
         print(traceback.format_exc())
 
 if not url:
-    print('Fail in login')
+    print('Fail in login, terminating...')
     exit(1)
 
 r2 = bjySession.get('https://m.bjyouth.net/dxx/my-integral?type=2&page=1&limit=15')
-res = json.loads(r2.text)
-if f"学习课程：《{title}》" in list(map(lambda x: x['text'], res['data'])):
+have_learned = json.loads(r2.text)
+if f"学习课程：《{title}》" in list(map(lambda x: x['text'], have_learned['data'])):
     print(f'{title} 在运行前已完成')
     exit(0)
 
@@ -98,22 +73,21 @@ def encrypt(t):
     exit(1)
 
 end_img_url = f'https://h5.cyol.com/special/daxuexi/{result.group(1)}/images/end.jpg'
-study_url = f"https://m.bjyouth.net/dxx/check?id={r['newCourse']['id']}&org_id={org_id}"
+study_url = f"https://m.bjyouth.net/dxx/check?id={course_id}&org_id={org_id}"
 
 r = bjySession.get(study_url)
-
 if r.text:
     print(
         f'Unexpected response: {r.text}'
     )
     exit(1)
 
 r = bjySession.get('https://m.bjyouth.net/dxx/my-integral?type=2&page=1&limit=15')
-res = json.loads(r.text)
-if f"学习课程：《{title}》" in list(map(lambda x: x['text'], res['data'])):
+have_learned = json.loads(r.text)
+if f"学习课程：《{title}》" in list(map(lambda x: x['text'], have_learned['data'])):
     print(f'{title} 成功完成学习')
     exit(0)
 else:
-    print(f'Seem finished {title}, but not confirmed as {f"学习课程：《{title}》"}' not in {
-        list(map(lambda x: x['text'], res['data']))})
+    print(f'完成{title}, 但未在检查中确认 {f"学习课程：《{title}》"}' not in {
+        list(map(lambda x: x['text'], have_learned['data']))})
     exit(1)

readme.md

@@ -22,7 +22,7 @@
 
 ​		[如何添加SECRET](https://docs.github.com/cn/actions/security-guides/encrypted-secrets#creating-encrypted-secrets-for-a-repository)
 
-3. 在Actions界面**手动启用(默认被禁用)** Workflows，**DaXueXi** 自动跟随本分支更新(以希望在有变化时不用再手动fetch upstream)，如有安全顾虑或需要修改等可选择没有自动更新的 **DaXueXi (No update)**
+3. 在Actions界面**手动启用(默认被禁用)** Workflows，**DaXueXi** 自动跟随本分支更新(以希望在有变化时不用再手动fetch upstream)，如有安全顾虑**或需要修改**等可选择没有自动更新的 **DaXueXi (No update)**
 4. (可以手动运行一次试验)，可以在Run python中看到打印的结果信息
 
 # 其他

utility.py

@@ -0,0 +1,59 @@
+from base64 import b64encode
+from io import BytesIO
+
+from Crypto.Cipher import PKCS1_v1_5 as Cipher_pksc1_v1_5
+from Crypto.PublicKey import RSA
+from PIL import Image
+from ddddocr import DdddOcr
+
+
+def encrypt(t):
+    public_key = "-----BEGIN PUBLIC KEY-----\nMIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQD5uIDebA2qU746e/NVPiQSBA0Q3J8/G23zfrwMz4qoip1vuKaVZykuMtsAkCJFZhEcmuaOVl8nAor7cz/KZe8ZCNInbXp2kUQNjJiOPwEhkGiVvxvU5V5vCK4mzGZhhawF5cI/pw2GJDSKbXK05YHXVtOAmg17zB1iJf+ie28TbwIDAQAB\n-----END PUBLIC KEY-----"
+    rsa_key = RSA.importKey(public_key)
+    cipher = Cipher_pksc1_v1_5.new(rsa_key)
+    cipher_text = b64encode(cipher.encrypt(t.encode()))
+    return cipher_text.decode()
+
+
+ocr = DdddOcr()
+def cap_recognize(cap):
+    return ocr.classification(denoise(cap))
+
+
+def denoise(cap):
+    img = Image.open(BytesIO(cap))
+    _STEPS_LAYER_1 = ((1, 1), (1, 0), (1, -1), (0, 1), (0, -1), (-1, 1), (-1, 0), (-1, -1))
+    STEPS8 = _STEPS_LAYER_1
+    _PX_WHITE = (250, 250, 250)
+
+    def _denoise(img, steps, threshold, repeat):
+        for _ in range(repeat):
+            for j in range(img.width):
+                for i in range(img.height):
+                    px = img.getpixel((j, i))
+                    if px == _PX_WHITE:  # 自身白
+                        continue
+                    count = 0
+                    if px[0] < px[1] + px[2]:
+                        count = 2
+                    for x, y in steps:
+                        j2 = j + y
+                        i2 = i + x
+                        if 0 <= j2 < img.width and 0 <= i2 < img.height:  # 边界内
+                            if img.getpixel((j2, i2)) == _PX_WHITE:  # 周围白
+                                count += 1
+                        else:  # 边界外全部视为黑
+                            count += 1
+                    if count >= threshold:
+                        img.putpixel((j, i), _PX_WHITE)
+
+        return img
+
+    def denoise8(img, steps=STEPS8, threshold=7, repeat=2):
+        """ 考虑外一周的降噪 """
+        return _denoise(img, steps, threshold, repeat)
+
+    buf = BytesIO()
+    denoise8(img).save(buf, format='PNG')
+    byte_im = buf.getvalue()
+    return byte_im