Lasse Studion
2 years ago
commit
4379fe083e
2 changed files with 199 additions and 0 deletions
@ -0,0 +1,3 @@ |
||||
* |
||||
!actapublica.py |
||||
!.gitignore |
||||
@ -0,0 +1,196 @@ |
||||
# 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) |
||||
Loading…
Reference in new issue