actapublica_api/actapublica.py

# Testa om det finns en giltig access token och skapa den om den inte finns,
# spara till access_token.json
import json
from datetime import datetime, timedelta
from pprint import pprint
from time import time
from urllib.request import Request, urlopen

import requests

# API documentation: https://actapublica.docs.apiary.io/#reference/0/search-results-of-users-agents/post


class ActaPublica:
    def __init__(self, client_id=False, client_key=False):
        if not client_id:
            client_id = input("Client ID:")
            client_key: input("Client key:")
        self.token = self.get_token(client_id, client_key)
        self.params = self.reset_params()
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {self.token}",
        }
        self.menu()

    def menu(self):
        # What to search for.
        print(
            """
              Vad vill du söka efter? Vill du göra en snabb sökning skriv bara sökorden.
              1. Personnummer
              A. Jag är klar, sök i Acta Publica.
              B. Inget, avsluta programmet.
              """
        )
        user_input = input(">> ").strip()

        if user_input == "A":
            self.search()
        elif user_input == "B":
            print("Avslutar")
            exit()
        elif user_input == "1":
            self.add_pnr()
        else:
            self.quick_search(user_input)

        # TODO Add more alternatives, and parameters like where to search etc.
        # # Where to search
        # user_input = input('>> ')
        # print('''
        #     Var vill du söka?
        #     1. Domstolar
        #     ''')

        # if user_input == '':
        #     pass # Search everywhere.
        # elif user_input == '1':

    def search(self):
        """Do a search in Acta Publica."""

        request = Request(
            "https://api.arkivet.actapublica.se/search",
            data=bytes(json.dumps(self.params), encoding="utf-8"),
            headers=self.headers,
        )
        r = urlopen(request).read()  # Comes in bytes. #* Change to .json()?

        self.search_result = json.loads(r)  # Data as a dict.

    def quick_search(self, query):
        """Do a quick search."""
        self.reset_params()
        self.params["query"] = str(query)
        self.search()

    def view_result(self):
        for result in self.search_result:
            pprint(result, "\n---\n")

    def download_documents(self):
        for i in range(0, len(self.search_result["hits"])):
            url = self.search_result["hits"][i]["document_download_link"]
            filnamn = self.search_result["hits"][i]["filename"]

            pdf = requests.get(url, allow_redirects=True)
            with open(filnamn, "wb") as f:
                f.write(pdf.content)

    def add_pnr(self, condition="OR"):
        user_input = input("Personnummer: ")
        if "csv" in user_input:
            with open(user_input) as csv:
                l = [i.strip() for i in csv]
        else:
            l = user_input.split(",")

        # Remove spaces.
        l = [i.strip() for i in l]

        self.params["personnummer"] = {"condition": condition, "values": [l]}
    def new_token(client_id, client_key):
        """
        Get an access token from Acta Publica.
        """
        values = {
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_key,
            "scope": "document.view,document.list,agent.detail,agent.list,document.download",
        }

        headers = {"Content-Type": "application/json", "accept": "application/json"}

        request = Request(
            "https://api.arkivet.actapublica.se/authorize",
            data=bytes(json.dumps(values), encoding="utf-8"),
            headers=headers,
        )

        r = urlopen(request).read()
        parsed_json = json.loads(r.decode("utf8"))

        parsed_json["tid"] = str(datetime.now())
        with open("access_token.json", "w") as f:
            json.dump(parsed_json, f)
        return parsed_json
    
        #   This is a Python function that generates an access token for Acta Publica's API. It takes two parameters: `client_id` and `client_key`, which are used to authenticate the request. The function returns a dictionary containing the access token and other relevant information.
        # Here is a breakdown of the code:
        # 1. The function starts by defining a dictionary called `values`. This dictionary will contain the parameters needed to generate the access token.
        # 2. The function then defines a dictionary called `headers`. This dictionary will contain the headers needed for the HTTP request.
        # 3. The function then creates an instance of the `Request` class from the `urllib.request` module. This object represents the HTTP request that will be sent to Acta Publica's API.
        # 4. The function then sets the `data` parameter of the `Request` object to a JSON-encoded string containing the parameters needed to generate the access token.
        # 5. The function then sets the `headers` parameter of the `Request` object to a dictionary containing the headers needed for the HTTP request.
        # 6. The function then sends the HTTP request to Acta Publica's API using the `urlopen` function from the `urllib.request` module. This function returns an instance of the `Response` class, which represents the response received from Acta Publica's API.
        # 7. The function then reads the contents of the response using the `read()` method and stores them in a variable called `r`.
        # 8. The function then decodes the contents of the response from UTF-8 encoding using the `decode()` method and stores them in a variable called `parsed_json`.
        # 9. The function then extracts the access token from the response and stores it in a variable called `access_token`.
        # 10. The function then updates the dictionary `parsed_json` with the access token and other relevant information.
        # 11. The function then writes the updated dictionary to a JSON file called `access_token.json` using the `dump()` method from the `json` module.
        # 12. The function then returns the updated dictionary containing the access token and other relevant information.

    def get_token(self, client_id, client_key):
        """Look for access token and create one if not local available."""
        try:
            with open("access_token.json") as f:
                token_data = json.load(f)
            token = token_data["access_token"]
            valid = timedelta(seconds=token_data["expires_in"])
            skapad = datetime.fromisoformat(token_data["tid"])
            expires = skapad + valid
            if (skapad + valid) - datetime.now() > timedelta(seconds=3600):
                print(
                    "\nAccess token {} giltig till {} {}".format(
                        token[:5] + "...",
                        datetime.date(skapad + valid),
                        str(datetime.time(skapad + valid))[:5],
                    )
                )

            else:
                token = self.new_token(client_id, client_key)["access_token"]
                print(
                    "\nAccess token {} skapad {}".format(
                        token[:5] + "...", datetime.now()
                    )
                )

        except FileNotFoundError:
            token = self.new_token(client_id, client_key)["access_token"]
            print(
                "\nAccess token {} skapad {}".format(token[:5] + "...", datetime.now())
            )

        return token

    def reset_params(self):
        """Reset parameters."""
        parametrar = {
            "size": "1000",
            "start_date": "01-01-2001 00:00:00",
            "end_date": time.strftime("%d-%m-%Y %H:%M:%S"),
            "sort": "",
        }


if __name__ == "__main__":
    ap = ActaPublica()

    while True:
        ap.menu()
        pprint(ap.search_result)