axiom-team
/
jaklis
7
2
Fork 3
Code Issues 4 Pull Requests Releases Wiki Activity

refacto #1

This commit is contained in:
poka 2023-09-10 01:53:38 +02:00
parent e188d4da94
commit 595f7722bf
2 changed files with 64 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
lib/geolocProfiles.py
View File

@ -1,12 +1,12 @@
import requests, sys import requests
from time import time from time import time
from lib.cesiumCommon import CesiumCommon from lib.cesiumCommon import CesiumCommon
from lib.gvaWallets import ListWallets from lib.gvaWallets import ListWallets
class GeolocProfiles(CesiumCommon): class GeolocProfiles(CesiumCommon):
# Configure JSON document POST ready to send
def getCesiumProfiles(self): def getCesiumProfiles(self):
# Send a POST request to the Cesium profiles API
response = requests.post( response = requests.post(
"https://g1.data.e-is.pro/user/profile/_search?scroll=2m", "https://g1.data.e-is.pro/user/profile/_search?scroll=2m",
json={ json={
@ -45,24 +45,24 @@ class GeolocProfiles(CesiumCommon):
finalResult = response.json()["hits"]["hits"] finalResult = response.json()["hits"]["hits"]
while True: while True:
# Effectuez une requête de défilement pour obtenir la page suivante # Send a scroll request to get the next page
response_scroll = requests.post( response_scroll = requests.post(
"https://g1.data.e-is.pro/_search/scroll", "https://g1.data.e-is.pro/_search/scroll",
json={"scroll_id": scroll_id, "scroll": "2m"}, json={"scroll_id": scroll_id, "scroll": "2m"},
) )
# Vérifiez si la réponse est vide (aucun résultat) ou si la durée du défilement est écoulée # Check if the response is empty (no results) or if there's an error
if ( if (
not response_scroll.json()["hits"]["hits"] not response_scroll.json()["hits"]["hits"]
or "error" in response_scroll.json() or "error" in response_scroll.json()
): ):
break break
else: else:
finalResult = finalResult + response_scroll.json()["hits"]["hits"] finalResult.extend(response_scroll.json()["hits"]["hits"])
# Traitez les résultats ici # Process the results here
# Une fois terminé, supprimez le contexte de défilement # Delete the scroll context when done
requests.delete( requests.delete(
"https://g1.data.e-is.pro/_search/scroll", json={"scroll_id": [scroll_id]} "https://g1.data.e-is.pro/_search/scroll", json={"scroll_id": [scroll_id]}
) )
@ -70,7 +70,8 @@ class GeolocProfiles(CesiumCommon):
return finalResult return finalResult
def getGVAProfiles(self, node): def getGVAProfiles(self, node):
gva = ListWallets(node, False, False, False, False, True) # Retrieve GVA profiles using the ListWallets class
gva = ListWallets(node, map=True)
return gva.sendDoc() return gva.sendDoc()
def formatProfiles(self, cesiumProfiles, gvaProfiles): def formatProfiles(self, cesiumProfiles, gvaProfiles):
@ -78,52 +79,20 @@ class GeolocProfiles(CesiumCommon):
for profile in cesiumProfiles: for profile in cesiumProfiles:
source = profile["_source"] source = profile["_source"]
pubkey = profile["_id"] pubkey = profile["_id"]
# if pubkey == "JEGCJSfKbFyxUnnHBQUqe9U4fhMMpdWvr7zdU56EZHsa":
# print(pubkey)
# sys.exit(0)
if pubkey not in gvaProfiles: if pubkey not in gvaProfiles:
continue continue
try: # Extract necessary information from the profiles
isMember = gvaProfiles[pubkey]["id"]["isMember"] id_info = gvaProfiles[pubkey].get("id") or {}
except: isMember = id_info.get("isMember", False)
isMember = False userId = id_info.get("username")
title = source.get("title")
try: city = source.get("city")
userId = gvaProfiles[pubkey]["id"]["username"] avatar = source.get("avatar")
except: socials = source.get("socials")
userId = None description = source.get("description")
address = source.get("address")
try:
title = source["title"]
except:
title = None
try:
city = source["city"]
except:
city = None
try:
avatar = source["avatar"]
except:
avatar: None
try:
socials = source["socials"]
except:
socials = None
try:
description = source["description"]
except:
description = None
try:
address = source["address"]
except:
address = None
walletsResult.append( walletsResult.append(
{ {
@ -140,6 +109,4 @@ class GeolocProfiles(CesiumCommon):
} }
) )
finalResult = {"wallets": walletsResult, "time": int(time())} return {"wallets": walletsResult, "time": int(time())}
return finalResult

91
lib/gvaWallets.py
View File

@ -1,28 +1,29 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import sys, re, os.path, json, ast import sys
from termcolor import colored import json
from lib.natools import fmt, sign, get_privkey
from gql import gql, Client from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport from gql.transport.aiohttp import AIOHTTPTransport
from lib.natools import fmt, sign, get_privkey
PUBKEY_REGEX = "(?![OIl])[1-9A-Za-z]{42,45}"
class ListWallets: class ListWallets:
def __init__(self, node, brut, mbr, nonMbr, larf, map=False): def __init__(
self.mbr = mbr self, node=False, brut=False, mbr=False, nonMbr=False, larf=False, map=False
self.larf = larf ):
self.nonMbr = nonMbr # Initialize the ListWallets class with optional filters
self.brut = brut self.mbr = mbr # Filter for members
self.map = map self.larf = larf # Filter for non-empty identities
self.nonMbr = nonMbr # Filter for non-members
self.brut = brut # Output format flag (brut or JSON)
self.map = map # Output format flag (map or list)
# Define Duniter GVA node # Define Duniter GVA node
transport = AIOHTTPTransport(url=node) transport = AIOHTTPTransport(url=node)
self.client = Client(transport=transport, fetch_schema_from_transport=True) self.client = Client(transport=transport, fetch_schema_from_transport=True)
def sendDoc(self): def sendDoc(self):
# Build wallets generation document # Define the GraphQL query to retrieve wallet information
queryBuild = gql( queryBuild = gql(
""" """
{ {
@ -46,58 +47,54 @@ class ListWallets:
} }
} }
} }
""" """
) )
# Send wallets document
try: try:
# Execute the GraphQL query
queryResult = self.client.execute(queryBuild) queryResult = self.client.execute(queryBuild)
except Exception as e: except Exception as e:
sys.stderr.write("Echec de récupération de la liste:\n" + str(e) + "\n") # Handle any exceptions that occur during the query
sys.stderr.write("Failed to retrieve the list:\n" + str(e) + "\n")
sys.exit(1) sys.exit(1)
jsonBrut = queryResult["wallets"]["edges"] jsonBrut = queryResult["wallets"]["edges"]
walletList = [] walletList = []
walletMap = {} walletMap = {}
for i, trans in enumerate(jsonBrut): for i, trans in enumerate(jsonBrut):
dataWork = trans["node"] dataWork = trans["node"]
if self.mbr and ( identity = dataWork["idty"]
dataWork["idty"] == None or dataWork["idty"]["isMember"] == False is_member = identity and identity["isMember"]
):
continue # Apply filters based on member status and larf flag
if self.nonMbr and ( member_filter = self.mbr and not is_member
dataWork["idty"] == None or dataWork["idty"]["isMember"] == True non_member_filter = self.nonMbr and is_member
): larf_filter = self.larf and identity
continue if member_filter or non_member_filter or larf_filter:
if self.larf and (dataWork["idty"] != None):
continue continue
wallet_data = {
"pubkey": dataWork["script"],
"balance": dataWork["balance"]["amount"] / 100,
"id": identity,
}
if self.map: if self.map:
walletMap[dataWork["script"]] = { walletMap[dataWork["script"]] = wallet_data
"balance": dataWork["balance"]["amount"] / 100,
"id": dataWork["idty"],
}
else: else:
walletList.append( walletList.append(wallet_data)
{
"pubkey": dataWork["script"],
"balance": dataWork["balance"]["amount"] / 100,
"id": dataWork["idty"],
}
)
if self.brut: if self.brut:
names = [] # Generate a list of formatted wallet names using list comprehension
for dataWork in walletList: names = [
if self.mbr or self.nonMbr: wallet["pubkey"]
names.append(dataWork["pubkey"] + " " + dataWork["id"]["username"]) if not (self.mbr or self.nonMbr) or wallet["id"] is None
else: else f'{wallet["pubkey"]} {wallet["id"]["username"]}'
names.append(dataWork["pubkey"]) for wallet in walletList
]
return "\n".join(names) return "\n".join(names)
else: else:
if self.map: # Return JSON data in either map or list format
return json.dumps(walletMap, indent=2) return json.dumps(walletMap if self.map else walletList, indent=2)
else:
return json.dumps(walletList, indent=2)