All/Python/local/ear-tag/fetcher.py

import json
import requests
import time


class Fetcher:
    def __init__(self, farmId: "str" = "330110004"):
        self.FarmId: "str" = farmId
        self.Session: "requests.Session" = requests.session()
        self.ids: "list[tuple]" = []
        self.data: "list[tuple]" = [("id", "mark", "timestamp", "time", "inner", "outer", "env", "act")]

    def login(self):
        url = "http://119.3.44.183:8016/admin/my/loginMultilevel"
        payload = {"accountName": "superadmin", "password": "hm123456", "farmId": self.FarmId}
        resp = self.Session.post(url=url, json=payload)
        assert resp.status_code == 200, f"login failed: {resp.status_code}"
        self.Session.headers.setdefault("Accesstoken", resp.json()["data"]["token"])

    def listIds(self):
        url = "http://119.3.44.183:8016/manage2/eartagRegister/listUserEartagData"
        payload = {"pageNo": 1, "pageSize": 20, "earmark": "", "status": "", "farmId": "330110004"}
        for i in range(133):  # 2624 = 131 * 20 + 4
            payload["pageNo"] = i + 1
            try:
                resp = self.Session.post(url=url, json=payload)
                this = [(itr["earmark"], itr["id"]) for itr in resp.json()["data"]["records"]]
                self.ids += this
                print(f"> page={i + 1}, count={len(this)}")
            except Exception as e:
                print(f"> ERROR: page={i + i}\n", e)
            time.sleep(0.5)
            break  # first 20 only
        fp = open("raw-data/ear-tags.json", "w", encoding="utf-8")
        json.dump(self.ids, fp)
        fp.close()
        print(f"> got {len(self.ids)} ids in raw-data/ear-tags.json")

    @staticmethod
    def timestamp(ts: "str") -> "int":
        return int(time.mktime(time.strptime(ts, "%Y-%m-%d %H:%M:%S")))

    @staticmethod
    def notNull(data):
        return data if data is not None else 0

    def clean(self, data: "list[dict]", cur: "tuple") -> "list[tuple]":
        return [(
            cur[1], cur[0], self.timestamp(itr["addTime"]), itr["addTime"], self.notNull(itr["earTemp1"]),
            self.notNull(itr["earTemp2"]), self.notNull(itr["envTemp1"]), self.notNull(itr["act"])
        ) for itr in data]

    def fetch_one(self, tag: "str", start: "str", end: "str", _id: "int" = -1) -> "list[tuple]":
        url = "http://119.3.44.183:8016/manage2/eartagData/getEnvByTime"
        payload = {"earmark": tag, "startDate": start, "endDate": end, "farmId": self.FarmId}
        resp = self.Session.post(url=url, json=payload)
        assert resp.status_code == 200, f"> ERROR: fetch <{tag}> failed!"
        return self.clean(resp.json()["data"], (_id, tag))

    def fetch_all(self):
        total = len(self.ids)
        for (index, current) in enumerate(self.ids):
            try:
                this = self.fetch_one(current[1], "2023-11-14 00:00:00", "2024-02-08 00:00:00", current[0])
                self.data += this
                print(f"> process [{index + 1:02d}/{total}], mark={current[0]}, id={current[1]}, count={len(this)}")
            except Exception as e:
                print(f"> ERROR: mark={current[0]}, id={current[1]}\n", e)
            time.sleep(1)
        self.csv()

    @staticmethod
    def tuple2str(t: "tuple") -> "str":
        return f"{t[0]},{t[1]},{t[2]},{t[3]},{t[4]},{t[5]},{t[6]},{t[7]}"

    def csv(self):
        mid = [self.tuple2str(itr) for itr in self.data]
        data = "\n".join(mid)
        fp = open("raw-data/data.csv", "w", encoding="utf-8")
        fp.write(data)
        fp.close()
        print(f"> all data saved in raw-data/data.csv")

    def run(self):
        self.login()
        self.listIds()
        self.fetch_all()


if __name__ == "__main__":
    fetch = Fetcher()
    fetch.run()