Merge pull request #28 from roleecorn/dev

Dev

Author: roleecorn

app.py

@@ -1,4 +1,5 @@
-from flask import Flask, render_template, jsonify, request
+from flask import Flask, render_template, jsonify
+from flask import send_from_directory
 import flask
 import time
 import re
@@ -51,6 +52,11 @@ def start_cite():
     return jsonify(redirect_url='/src/index.html')
 
 
+@app.route('/favicon.ico')
+def favicon():
+    return send_from_directory(home / 'src', 'web.ico')
+
+
 @app.route('/run_test')
 def run_test():
     thread = threading.Thread(target=background_task, args=('test',))
@@ -77,6 +83,8 @@ def run_ocr():
     thread = threading.Thread(target=background_task, args=('ocr',))
     thread.start()
     return jsonify(message="進行圖片光學解析")
+
+
 # run_ocr
 if __name__ == '__main__':
     app.run(debug=False)

cite_envs/eddiebauer.yml

@@ -1,8 +1,7 @@
 lastupdate: 2023-09-08
 target: tile_wrapper_outer
-nextpage: 
-  item: load_more
-  method: extend
+nextpage: load_more
+method: append
 position:
   money: 
     X: 116

cite_envs/sample.yml

@@ -1,19 +1,20 @@
-lastupdate: '2023-09-30 14:23:57.318304'
-nextpage: ''
+lastupdate: 2023-09-08
+target: tile_wrapper_outer
+nextpage: load_more
+method: append
 position:
-  money:
-    X: 169
-    Y: 200
-    h: -18
-    w: -166
-  star:
-    X: 0
-    Y: 0
-    h: 111
-    w: 0
-  title:
+  money: 
+    X: 116
+    Y: 460
+    w: -114
+    h: -14
+  title: 
     X: 4
     Y: 424
-    h: 38
     w: 332
-target: tile_wrapper_outer
+    h: 38
+  star: 
+    X: 116
+    Y: 485
+    w: -27
+    h: -22

classversion.py

@@ -2,84 +2,57 @@
 import sqlite3
 from pathlib import Path
 from datetime import datetime
-import os
 import time
-
+from typing import List
 # Third-party imports
 from dotenv import load_dotenv
 from selenium.common.exceptions import NoSuchElementException
 from selenium.common.exceptions import ElementNotInteractableException
 import yaml
+from views.selenium_drive import driver_class
+from views.new_db import sql_add
 
 # Load environment variables (special case)
 load_dotenv()
 
 # Local application imports that depend on environment variables
-import util
-import driver_control
 from debugger import Debugger
+import driver_control
+import util
+
 
 
-class ocr_crawler:
+class ocr_crawler(driver_class):
     def __init__(self, cite: str, test: bool = True) -> None:
+        super().__init__()
         self.cite = cite
         self.home = Path.cwd()
         self.test = test
-        # self.driverpath = self.home / 'chromedriver.exe'
-        self.driverpath = Path(os.getenv("DriverPath"))
-        self.chromepath = Path(os.getenv("Chromepath"))
-        self.driver = None
         self.db_path = self.home / 'sql' / f"{cite}.db"
         self.read_csv()
-        self.current_date = str(datetime.now().date())
+        current_time = datetime.now()
+        self.current_date = str(current_time.date())
+        self.version = int(current_time.timestamp())
         if not self.db_path.exists():
-            self.sql_add()
+            Debugger.info_print("新增資料庫")
+            sql_add(db_path=self.db_path, cite=cite)
             time.sleep(1)
         with open(self.home / 'cite_envs' / f"{cite}.yml", 'r') as file:
             cite_config = yaml.safe_load(file)
         self.target_class: str = cite_config.get('target', "")
         self.position: dict[str, dict[str, int]
                             ] = cite_config.get('position', {})
-        self.nextpage: dict[str, str] = cite_config.get('nextpage', {})
-        # self.SESSION = sqlite3.connect(self.db_path)
-        self.version = int(time.time())
-
-    def sql_add(self) -> None:
-        Debugger.info_print('new database')
-        status = sqlite3.connect(self.db_path)
-        query = f"""
-        CREATE TABLE {self.cite} (
-            name TEXT,
-            imgcode TEXT,
-            oriprice INT,
-            price INT,
-            color INT,
-            feature TEXT,
-            gender INT,
-            brand TEXT,
-            fabric TEXT,
-            path TEXT,
-            ver INT
-        )
-        """
-        status.execute(query)
-        status.close()
+        self.nextpage: str = cite_config.get('nextpage', "")
+        self.nextpage_method: str = cite_config.get('method', "")
 
     def read_csv(self) -> None:
         self.listsite, self.site_feature = util.read_csv(
             (self.home / 'cite_file' / f'{self.cite}_test.csv'))
 
-    def new_driver(self):
-        Debugger.info_print('new driver')
-        self.driver = util.new_driver(dpath=self.driverpath,
-                                      cpath=self.chromepath)
-
-    def close(self):
-        if self.driver:
-            self.driver.close()
-
-    def one_page_start(self, imgpath: Path, img_features: list[str],
-                       ocr: bool = False):
+    def one_page_capture(self, imgpath: Path,
+                         img_features: List[str],
+                         ocr: bool = False,
+                         save: bool = True):
         """
         執行一個頁面的截圖
         """
@@ -90,94 +63,67 @@ def one_page_start(self, imgpath: Path, img_features: list[str],
                 "class name", self.target_class)
         except NoSuchElementException:
             Debugger.error_print('element not find in target_element')
+            return
         SESSION = sqlite3.connect(self.db_path)
         for element in target_elements:
             tmp = util.capture(ele=element, path=imgpath)
             datas = {
-                # "price": 100.50,
-                # "oriprice": 150.00,
                 "imgcode": tmp,
-                # "facturer": "ABC Company",
-                # "feature": "Waterproof",
-                # "color": "Blue",
-                # "name": "Cool Shoe",
-                # "star": 4.5,
                 "path": '/'.join(img_features),
-                # "sex": 1,
-                # # Assuming 0 for Female, 1 for Male, -1 for Unknown
                 "ver": self.version
             }
             if ocr:
-                datas["price"] = util.Ocr(img=imgpath/tmp,
-                                          posit=self.position['money'])
-                datas["name"] = util.Ocr(img=imgpath/tmp,
-                                         posit=self.position['title'])
-                datas["star"] = util.Ocr(img=imgpath/tmp,
-                                         posit=self.position['star'])
+                price = util.Ocr(img=imgpath/tmp,
+                                 posit=self.position['money'])
+                datas["price"] = util.remove_non_number(price)
+                name = util.Ocr(img=imgpath/tmp,
+                                posit=self.position['title'])
+                datas["name"] = util.remove_non_alphanumeric(name)
+                star = util.Ocr(img=imgpath/tmp,
+                                posit=self.position['star'])
+                datas["star"] = util.remove_non_number(star)
             clo = util.cloth(datas=datas)
             clo.writedb(db=SESSION, tablename=self.cite)
             del clo
-        SESSION.commit()
+        if save:
+            SESSION.commit()
+        else:
+            SESSION.rollback()
         SESSION.close()
 
     def test_start(self):
         """
         執行第一個網址的執行
         """
-        Debugger.info_print('test start')
+        Debugger.info_print('Test start')
         self.driver.get(url=self.listsite[0])
         time.sleep(3)
         if self.driver.current_url != self.listsite[0]:
             self.driver.get(url=self.listsite[0])
         p = self.home / 'image' / self.cite / self.current_date
         imgpath = util.check_imgpath(imgpath=p, imgfile=['test'])
-        self.one_page_start(imgpath=imgpath, img_features=['test'])
+        self.one_page_capture(imgpath=imgpath,
+                              img_features=['test'], save=False)
 
     def regular_start(self, subcite: int, ocr: bool = False):
         """
         執行選定目標編號網址
         """
         img_features = self.site_feature[subcite]
-        Debugger.info_print(f'regular start {img_features}')
         self.driver.get(url=self.listsite[subcite])
         if self.driver.current_url != self.listsite[subcite]:
             self.driver.get(url=self.listsite[subcite])
         p = self.home / 'image' / self.cite / self.current_date
         imgpath = util.check_imgpath(imgpath=p,
                                      imgfile=img_features)
-        if self.nextpage['method'] == 'extend':
-            while True:
-                driver_control.go_bottom_and_wait(driver=self.driver)
-                try:
-                    buttom = self.driver.find_element(
-                        "class name", self.nextpage['item'])
-                    self.driver.execute_script(
-                        "arguments[0].scrollIntoView();", buttom)
-                    buttom.click()
-                except NoSuchElementException:
-                    Debugger.info_print('no nextpage')
-                    break
-                except Exception as e:
-                    Debugger.error_print(str(e))
-                    break
-            self.one_page_start(
+        if self.nextpage_method == 'append':
+            driver_control.append_page(obj=self)
+            self.one_page_capture(
                 imgpath=imgpath, img_features=img_features, ocr=ocr)
-        elif self.nextpage['method'] == 'new':
-            while True:
-                self.one_page_start(
-                    imgpath=imgpath, img_features=img_features, ocr=ocr)
-                try:
-                    buttom = self.driver.find_element(
-                        "class name", self.nextpage['item'])
-                    self.driver.execute_script(
-                        "arguments[0].scrollIntoView();", buttom)
-                    buttom.click()
-                except NoSuchElementException:
-                    Debugger.info_print('no nextpage')
-                    break
-                except Exception as e:
-                    Debugger.error_print(str(e))
-                    break
+        elif self.nextpage_method == 'next':
+            driver_control.next_page(obj=self, imgpath=imgpath,
+                                     img_features=img_features,
+                                     ocr=ocr)
         else:
             raise AttributeError
 
@@ -206,10 +152,7 @@ def shot_all_classes(self):
             try:
                 ele = self.driver.find_element(
                     "class name", element)
-            except NoSuchElementException:
-                continue
-            except Exception as e:
-                # Debugger.error_print(str(e))
+            except Exception:
                 continue
             try:
                 util.capture(ele=ele, path=search_path, name=element)

driver_control/__init__.py

@@ -1,9 +1,11 @@
 from .scroll_to_bottom_and_wait import scroll_to_bottom_and_wait
 from .scroll_to_bottom_and_wait import go_bottom_and_wait
 from .get_all_class import get_all_classes
+from .nextpage import append_page, next_page
 
 __all__ = ["scroll_to_bottom_and_wait",
            "get_all_classes",
            "go_bottom_and_wait",
+           'append_page', 'next_page',
            ]
-__version__ = "0.2"
+__version__ = "0.3"

driver_control/get_all_class.py

@@ -1,7 +1,8 @@
 from selenium import webdriver
+from typing import List
 
 
-def get_all_classes(driver: webdriver.Chrome) -> list[str]:
+def get_all_classes(driver: webdriver.Chrome) -> List[str]:
     """
     Get all unique class names from the current page.
     :param driver: The selenium webdriver instance.

driver_control/nextpage.py

@@ -0,0 +1,42 @@
+from selenium.common.exceptions import NoSuchElementException
+from selenium.webdriver import ActionChains
+from debugger import Debugger
+
+
+def next_page(obj, imgpath, img_features, ocr):
+    while True:
+        obj.one_page_capture(
+            imgpath=imgpath, img_features=img_features, ocr=ocr)
+        try:
+            buttom = obj.driver.find_element(
+                "class name", obj.nextpage)
+            action = ActionChains(obj.driver)
+            action.move_to_element(buttom)
+            action.click().perform()
+            # obj.driver.execute_script(
+            #     "arguments[0].scrollIntoView();", buttom)
+            # buttom.click()
+        except NoSuchElementException:
+            break
+        except Exception as e:
+            Debugger.error_print(str(e))
+            break
+
+
+def append_page(obj):
+    while True:
+        try:
+            buttom = obj.driver.find_element(
+                "class name", obj.nextpage)
+            # obj.driver.execute_script(
+            #     "arguments[0].scrollIntoView();", buttom)
+            # buttom.click()
+            action = ActionChains(obj.driver)
+            action.move_to_element(buttom)
+            action.click().perform()
+        except NoSuchElementException:
+            Debugger.info_print('no nextpage')
+            break
+        except Exception as e:
+            Debugger.error_print(str(e))
+            break