aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--generalresearch/managers/thl/ipinfo.py748
-rw-r--r--generalresearch/managers/thl/payout.py223
-rw-r--r--generalresearch/managers/thl/userhealth.py158
-rw-r--r--generalresearch/managers/thl/wallet/__init__.py4
-rw-r--r--generalresearch/managers/thl/wallet/cashout_tasks.py3
-rw-r--r--generalresearch/managers/thl/wallet/tango.py41
-rw-r--r--generalresearch/managers/thl/wallet/user_payout.py392
-rw-r--r--generalresearch/models/thl/ipinfo.py133
-rw-r--r--generalresearch/models/thl/maxmind/__init__.py0
-rw-r--r--generalresearch/models/thl/maxmind/definitions.py22
-rw-r--r--generalresearch/models/thl/user_iphistory.py85
-rw-r--r--generalresearch/models/thl/wallet/cashout_method.py18
12 files changed, 560 insertions, 1267 deletions
diff --git a/generalresearch/managers/thl/ipinfo.py b/generalresearch/managers/thl/ipinfo.py
index 93914c3..575738b 100644
--- a/generalresearch/managers/thl/ipinfo.py
+++ b/generalresearch/managers/thl/ipinfo.py
@@ -1,736 +1,38 @@
-from __future__ import annotations
-
-import ipaddress
from collections.abc import Collection
-from decimal import Decimal
-from typing import TYPE_CHECKING
-
-import faker
-from grip_client.enums import AccessType
-from more_itertools import chunked
-from psycopg import Cursor
-from pydantic import PositiveInt
-
-from generalresearch.managers.base import (
- PostgresManager,
- PostgresManagerWithRedis,
-)
-from generalresearch.models.thl.ipinfo import (
- GeoIPInformation,
- IPGeoname,
- IPInformation,
- normalize_ip,
-)
-
-if TYPE_CHECKING:
- from generalresearch.models.custom_types import (
- CountryISOLike,
- IPvAnyAddressStr,
- )
- from generalresearch.pg_helper import PostgresConfig
-
-fake = faker.Faker()
-
-
-class IPGeonameManager(PostgresManager):
-
- def create_basic(
- self,
- geoname_id: PositiveInt,
- is_in_european_union: bool,
- country_iso: CountryISOLike,
- country_name: str,
- continent_code: str,
- continent_name: str,
- ) -> IPGeoname:
- instance = IPGeoname.model_validate(
- {
- "geoname_id": geoname_id,
- "country_iso": country_iso,
- "is_in_european_union": is_in_european_union,
- "country_name": country_name,
- "continent_code": continent_code,
- "continent_name": continent_name,
- }
- )
- self.pg_config.execute_write(
- query="""
- INSERT INTO thl_geoname (
- geoname_id, country_iso, is_in_european_union, country_name,
- continent_code, continent_name, updated
- )
- VALUES (
- %(geoname_id)s, %(country_iso)s, %(is_in_european_union)s,
- %(country_name)s, %(continent_code)s, %(continent_name)s,
- %(updated)s
- )
- ON CONFLICT (geoname_id) DO NOTHING;
- """,
- params=instance.model_dump(mode="json"),
- )
- return instance
-
- def create_or_update(self, ipgeo: IPGeoname):
- keys = list(ipgeo.model_fields.keys())
- data = ipgeo.model_dump_mysql()
-
- keys_str = ", ".join(keys)
- values_str = ", ".join([f"%({k})s" for k in keys])
- update_cols = set(keys) - {"geoname_id"}
- update_str = ", ".join([f"{k} = EXCLUDED.{k}" for k in update_cols])
-
- query = f"""
- INSERT INTO thl_geoname ({keys_str})
- VALUES ({values_str})
- ON CONFLICT (geoname_id)
- DO UPDATE SET {update_str}
- """
- self.pg_config.execute_write(query=query, params=data)
-
- def create(
- self,
- geoname_id: PositiveInt,
- continent_code: str,
- continent_name: str,
- country_iso: str | None,
- country_name: str | None = None,
- subdivision_1_iso: str | None = None,
- subdivision_1_name: str | None = None,
- subdivision_2_iso: str | None = None,
- subdivision_2_name: str | None = None,
- city_name: str | None = None,
- metro_code: int | None = None,
- time_zone: str | None = None,
- is_in_european_union: bool | None = None,
- ) -> IPGeoname:
-
- instance = IPGeoname.model_validate(
- {
- "geoname_id": geoname_id,
- "continent_code": continent_code,
- "continent_name": continent_name,
- "country_iso": country_iso,
- "country_name": country_name,
- "subdivision_1_iso": subdivision_1_iso,
- "subdivision_1_name": subdivision_1_name,
- "subdivision_2_iso": subdivision_2_iso,
- "subdivision_2_name": subdivision_2_name,
- "city_name": city_name,
- "metro_code": metro_code,
- "time_zone": time_zone,
- "is_in_european_union": is_in_european_union,
- }
- )
-
- self.pg_config.execute_write(
- query="""
- INSERT INTO thl_geoname
- ( geoname_id, continent_code, continent_name,
- country_iso, country_name,
- subdivision_1_iso, subdivision_1_name,
- subdivision_2_iso, subdivision_2_name,
- city_name, metro_code, time_zone, is_in_european_union,
- updated
- )
- VALUES (
- %(geoname_id)s, %(continent_code)s, %(continent_name)s,
- %(country_iso)s, %(country_name)s,
- %(subdivision_1_iso)s, %(subdivision_1_name)s,
- %(subdivision_2_iso)s, %(subdivision_2_name)s,
- %(city_name)s, %(metro_code)s, %(time_zone)s,
- %(is_in_european_union)s, %(updated)s
- )
- ON CONFLICT (geoname_id) DO NOTHING;
- """,
- params=instance.model_dump(mode="json"),
- )
-
- return instance
-
- def get_by_id(self, geoname_id: PositiveInt) -> IPGeoname:
- return self.fetch_geoname_ids(filter_ids=[geoname_id])[0]
-
- def fetch_geoname_ids(
- self,
- filter_ids: list[PositiveInt],
- ) -> list[IPGeoname]:
-
- if len(filter_ids) == 0:
- return []
-
- with self.pg_config.make_connection() as conn, conn.cursor() as c:
- res = []
- for chunk in chunked(filter_ids, 500):
- res.extend(
- self.fetch_geoname_ids_(
- c=c,
- filter_ids=chunk,
- )
- )
- return res
-
- def fetch_geoname_ids_(
- self,
- c: Cursor,
- filter_ids: list[PositiveInt],
- ) -> list[IPGeoname]:
-
- assert len(filter_ids) <= 500, "chunk me"
-
- c.execute(
- query="""
- SELECT g.geoname_id,
- g.continent_code, g.continent_name,
- g.country_iso, g.country_name,
- g.subdivision_1_iso, g.subdivision_1_name,
- g.subdivision_2_iso, g.subdivision_2_name,
- g.city_name, g.metro_code,
- g.time_zone, g.is_in_european_union,
- g.updated
- FROM thl_geoname AS g
- WHERE g.geoname_id = ANY(%s);
- """,
- params=[filter_ids],
- )
- return [IPGeoname.from_mysql(i) for i in c.fetchall()]
-
-
-class IPInformationManager(PostgresManager):
-
- def create_basic(
- self,
- ip: IPvAnyAddressStr,
- geoname_id: PositiveInt,
- country_iso: str,
- registered_country_iso: str,
- ) -> IPInformation:
- instance = IPInformation.model_validate(
- {
- "ip": ip,
- "geoname_id": geoname_id,
- "country_iso": country_iso,
- "registered_country_iso": registered_country_iso,
- }
- )
- instance.normalize_ip()
- self.pg_config.execute_write(
- query="""
- INSERT INTO thl_ipinformation
- (ip, country_iso, registered_country_iso, geoname_id, updated)
- VALUES (%(ip)s, %(country_iso)s, %(registered_country_iso)s,
- %(geoname_id)s, %(updated)s)
- ON CONFLICT (ip) DO NOTHING;
- """,
- params=instance.model_dump(mode="json"),
- )
- return instance
-
- def create(
- self,
- ip: IPvAnyAddressStr,
- geoname_id: PositiveInt | None = None,
- country_iso: str | None = None,
- registered_country_iso: str | None = None,
- is_anonymous: bool | None = None,
- is_anonymous_vpn: bool | None = None,
- is_hosting_provider: bool | None = None,
- is_public_proxy: bool | None = None,
- is_tor_exit_node: bool | None = None,
- is_residential_proxy: bool | None = None,
- autonomous_system_number: PositiveInt | None = None,
- autonomous_system_organization: str | None = None,
- domain: str | None = None,
- isp: str | None = None,
- mobile_country_code: str | None = None,
- mobile_network_code: str | None = None,
- network: str | None = None,
- organization: str | None = None,
- static_ip_score: float | None = None,
- user_type: AccessType | None = None,
- postal_code: str | None = None,
- latitude: Decimal | None = None,
- longitude: Decimal | None = None,
- accuracy_radius: int | None = None,
- ) -> IPInformation:
-
- instance = IPInformation.model_validate(
- {
- "ip": ip,
- "geoname_id": geoname_id,
- "country_iso": country_iso,
- "registered_country_iso": registered_country_iso,
- "is_anonymous": is_anonymous,
- "is_anonymous_vpn": is_anonymous_vpn,
- "is_hosting_provider": is_hosting_provider,
- "is_public_proxy": is_public_proxy,
- "is_tor_exit_node": is_tor_exit_node,
- "is_residential_proxy": is_residential_proxy,
- "autonomous_system_number": autonomous_system_number,
- "autonomous_system_organization": autonomous_system_organization,
- "domain": domain,
- "isp": isp,
- "mobile_country_code": mobile_country_code,
- "mobile_network_code": mobile_network_code,
- "network": network,
- "organization": organization,
- "static_ip_score": static_ip_score,
- "user_type": user_type,
- "postal_code": postal_code,
- "latitude": latitude,
- "longitude": longitude,
- "accuracy_radius": accuracy_radius,
- }
- )
- instance.normalize_ip()
-
- self.pg_config.execute_write(
- query="""
- INSERT INTO thl_ipinformation
- ( ip, geoname_id,
- country_iso, registered_country_iso,
- is_anonymous, is_anonymous_vpn,
- is_hosting_provider, is_public_proxy,
- is_tor_exit_node, is_residential_proxy,
- autonomous_system_number, autonomous_system_organization,
- domain, isp,
- mobile_country_code, mobile_network_code,
- network, organization, static_ip_score,
- user_type, postal_code, latitude, longitude,
- accuracy_radius,
- updated
- )
- VALUES (
- %(ip)s, %(geoname_id)s,
- %(country_iso)s, %(registered_country_iso)s,
- %(is_anonymous)s, %(is_anonymous_vpn)s,
- %(is_hosting_provider)s, %(is_public_proxy)s,
- %(is_tor_exit_node)s, %(is_residential_proxy)s,
- %(autonomous_system_number)s, %(autonomous_system_organization)s,
- %(domain)s, %(isp)s,
- %(mobile_country_code)s, %(mobile_network_code)s,
- %(network)s, %(organization)s, %(static_ip_score)s,
- %(user_type)s, %(postal_code)s, %(latitude)s, %(longitude)s,
- %(accuracy_radius)s,
- %(updated)s
- )
- ON CONFLICT (ip) DO NOTHING;
- """,
- params=instance.model_dump(mode="json"),
- )
-
- return instance
-
- def create_or_update(self, ipinfo: IPInformation):
- ipinfo.normalize_ip()
- keys = [key for key, field in ipinfo.model_fields.items() if not field.exclude]
- data = ipinfo.model_dump_mysql()
-
- keys_str = ", ".join(keys)
- values_str = ", ".join([f"%({k})s" for k in keys])
- update_cols = set(keys) - {"ip"}
- update_str = ", ".join([f"{k} = EXCLUDED.{k}" for k in update_cols])
-
- query = f"""
- INSERT INTO thl_ipinformation ({keys_str})
- VALUES ({values_str})
- ON CONFLICT (ip) DO UPDATE
- SET {update_str}
- """
- self.pg_config.execute_write(query, params=data)
-
- def get_ip_info(self, ip: IPvAnyAddressStr) -> IPInformation | None:
- res = self.fetch_ip_information(filter_ips=[ip])
- if len(res) != 1:
- return None
-
- return res[0]
-
- def fetch_ip_information(
- self,
- filter_ips: list[IPvAnyAddressStr],
- ) -> list[IPInformation]:
-
- if len(filter_ips) == 0:
- return []
-
- with self.pg_config.make_connection() as conn, conn.cursor() as c:
- res = []
- for chunk in chunked(filter_ips, 500):
- res.extend(
- self.fetch_ip_information_(
- c=c,
- filter_ips=chunk,
- )
- )
- return res
- def fetch_ip_information_(
- self,
- c: Cursor,
- filter_ips: list[IPvAnyAddressStr],
- ) -> list[IPInformation]:
- """
- IPs are converted to normalized form (/64 network exploded) for DB lookup,
- and are then matched back to the original queried form for return.
- e.g. '2600:1700:ece0:9410:055d:faf3:c15d:06e4' is passed in,
- it gets converted to '2600:1700:ece0:9410:0000:0000:0000:0000' for db lookup,
- the result gets
- """
+from grip_client import GRIPMMDBClient
- assert len(filter_ips) <= 500, "chunk me"
- normalized_ip_lookup = {ip: normalize_ip(ip)[0] for ip in filter_ips}
- normalized_ips = set(normalized_ip_lookup.values())
+from generalresearch.managers.base import Manager
+from generalresearch.models.custom_types import IPvAnyAddressStr
+from generalresearch.models.thl.ipinfo import GeoIPInformation
- c.execute(
- query="""
- SELECT i.ip, i.geoname_id,
- i.country_iso, i.registered_country_iso,
- i.is_anonymous, i.is_anonymous_vpn, i.is_hosting_provider,
- i.is_public_proxy, i.is_tor_exit_node, i.is_residential_proxy,
- i.autonomous_system_number, i.autonomous_system_organization,
- i.domain, i.isp,
- i.mobile_country_code, i.mobile_network_code,
- i.network, i.organization,
- i.static_ip_score, i.user_type, i.postal_code,
- i.latitude, i.longitude, i.accuracy_radius,
- i.updated
- FROM thl_ipinformation AS i
- WHERE i.ip = ANY(%s)
- """,
- params=[list(normalized_ips)],
- )
+class GeoIpInfoManager(Manager):
+ def __init__(self, grip_token: str, **kwargs) -> None:
+ self.grip_mmdb = GRIPMMDBClient(token=grip_token, **kwargs)
- return [IPInformation.from_mysql(i) for i in c.fetchall()]
-
- @staticmethod
- def monitor_ipinformation(pg_config: PostgresConfig) -> None:
- """Continually check our IPInformation table to ensure location information
- is being saved properly"""
-
- # Check the percentage of IPs that don't have a country_iso in the past 12 hours
- # Split query is 1000x faster
- query = """
- SELECT COUNT(*) AS numerator
- FROM thl_ipinformation
- WHERE updated >= NOW() - INTERVAL '12 hours'
- AND country_iso IS NULL;
- """
- # numerator = list(pg_config.execute_sql_query(query=query))[0]["numerator"]
-
- query = """
- SELECT COUNT(1) AS denominator
- FROM thl_ipinformation
- WHERE updated >= NOW() - INTERVAL '12 hours'
- """
- denominator = next(iter(pg_config.execute_sql_query(query=query)))[
- "denominator"
- ]
- if denominator == 0:
- pass
-
- # percent_empty = numerator / (denominator or 1)
- # TODO: Post to telegraf / grafana
-
-
-class GeoIpInfoManager(PostgresManagerWithRedis):
-
- def get(self, ip_address: IPvAnyAddressStr) -> GeoIPInformation | None:
- res = self.get_cache(ip_address)
- if res:
- return res
- res = self.get_mysql_if_exists(ip_address)
- if res:
- self.set_cache(res)
- return res
+ def get(self, ip_address: IPvAnyAddressStr) -> GeoIPInformation:
+ d = self.get_multi({ip_address})
+ return d[ip_address]
def get_multi(
self, ip_addresses: Collection[IPvAnyAddressStr]
- ) -> dict[IPvAnyAddressStr, GeoIPInformation | None]:
+ ) -> dict[IPvAnyAddressStr, GeoIPInformation]:
if not ip_addresses:
return {}
- # To deploy this, we still have (for the next 28 days) users who's
- # ipv6 history was looked up and saved using the full /128. We need
- # to pull those if the /64 doesn't exist.
- # See notes in get_cache_multi & get_mysql_multi
- res = self.get_cache_multi(ip_addresses=ip_addresses)
- missing_ips = {k for k, v in res.items() if v is None and k in ip_addresses}
- res_mysql = self.get_mysql_multi(ips=missing_ips)
- self.set_cache_multi({k: v for k, v in res_mysql.items() if v})
- res.update(res_mysql)
- return res
-
- def set_cache_multi(
- self, ipinfo_map: dict[IPvAnyAddressStr, GeoIPInformation]
- ) -> None:
- """Set multiple GeoIPInformation objects in Redis in one call."""
- if not ipinfo_map:
- return
-
- pipe = self.redis_client.pipeline(transaction=False)
- expire_seconds = 3 * 24 * 3600
- for ip, ipinfo in ipinfo_map.items():
- pipe.set(
- self.get_cache_key(ip),
- ipinfo.model_dump_json(),
- ex=expire_seconds,
- )
- pipe.execute()
-
- @staticmethod
- def compress_ip(ip: str) -> str:
- """
- To support looking up an ip in the db before we switched
- to using the exploded form. (remove me 28 days after 2025-11-15)
- """
- addr = ipaddress.ip_address(ip)
- if addr.version == 4:
- return str(addr)
- return addr.compressed
-
- def get_cache_multi(
- self, ip_addresses: Collection[IPvAnyAddressStr]
- ) -> dict[IPvAnyAddressStr, GeoIPInformation | None]:
- """Get multiple GeoIPInformation objects from Redis in one call.
-
- Returns a dict mapping IP address -> GeoIPInformation (or None if not in cache).
- """
- if not ip_addresses:
- return {}
- # We must do it like this b/c we could have multiple /128 ips that normalize
- # to the same normalized /64 ip, and we don't want to "loose" them.
- ip_norm_lookup = {ip: normalize_ip(ip) for ip in ip_addresses}
- normalized_ips = {v[0] for v in ip_norm_lookup.values()}
- # also lookup exact matches (can remove this 28 days from 2025-11-15)
- normalized_ips.update(ip_addresses)
- # also lookup compressed form ... (remove me also)
- normalized_ips.update({self.compress_ip(ip) for ip in ip_addresses})
-
- keys = [self.get_cache_key(ip) for ip in normalized_ips]
- res = self.redis_client.mget(keys)
- res = [GeoIPInformation.model_validate_json(raw) for raw in res if raw]
- gs = {x.ip: x for x in res}
-
- res2 = {}
- for ip, (normalized_ip, lookup_prefix) in ip_norm_lookup.items():
- if normalized_ip not in gs:
- # try the non-normalized (remove me also 28 days from 2025-11-15)
- if ip in gs:
- res2[ip] = gs[ip].model_copy()
- continue
- res2[ip] = None
- continue
- g = gs[normalized_ip]
- g.ip = ip
- g.lookup_prefix = lookup_prefix
- res2[g.ip] = g.model_copy()
- return res2
-
- def get_cache_key(self, ip_address: IPvAnyAddressStr) -> str:
- return self.cache_prefix + f"thl:GeoIpInfoManager:{ip_address}"
-
- def clear_cache(self, ip_address: IPvAnyAddressStr) -> None:
- # typically for testing
- self.redis_client.delete(self.get_cache_key(ip_address=ip_address))
-
- def set_cache(self, ipinfo: GeoIPInformation):
- ipinfo = ipinfo.model_copy()
- ipinfo.normalize_ip()
- data = ipinfo.model_dump_json()
- return self.redis_client.set(
- self.get_cache_key(ip_address=ipinfo.ip), data, ex=3 * 24 * 3600
- )
-
- def get_cache(self, ip_address: IPvAnyAddressStr) -> GeoIPInformation | None:
- normalized_ip, lookup_prefix = normalize_ip(ip_address)
- res: str = self.get_cache_raw(normalized_ip)
- if not res:
- return None
- g = GeoIPInformation.model_validate_json(res)
- g.ip = ip_address
- g.lookup_prefix = lookup_prefix
- return g
-
- def get_cache_raw(self, ip_address: IPvAnyAddressStr) -> str:
- return self.redis_client.get(self.get_cache_key(ip_address=ip_address))
-
- def get_mysql_if_exists(self, ip_address: IPvAnyAddressStr):
- try:
- return self.get_mysql(ip_address=ip_address)
- except AssertionError:
- return None
-
- def get_mysql_raw(self, ip_address: IPvAnyAddressStr):
- query = """
- SELECT
- geo.geoname_id,
- geo.continent_name,
- LOWER(geo.continent_code) AS continent_code,
- geo.country_name,
- LOWER(geo.country_iso) AS geo_country_iso,
- geo.subdivision_1_iso,
- geo.subdivision_1_name,
- geo.subdivision_2_iso,
- geo.subdivision_2_name,
- geo.city_name,
- geo.metro_code,
- geo.time_zone,
- geo.is_in_european_union,
- LOWER(ipinfo.country_iso) AS country_iso,
- ipinfo.registered_country_iso,
- ipinfo.is_anonymous,
- ipinfo.is_anonymous_vpn,
- ipinfo.is_hosting_provider,
- ipinfo.is_public_proxy,
- ipinfo.is_tor_exit_node,
- ipinfo.is_residential_proxy,
- ipinfo.autonomous_system_number,
- ipinfo.autonomous_system_organization,
- ipinfo.domain,
- ipinfo.isp,
- ipinfo.mobile_country_code,
- ipinfo.mobile_network_code,
- ipinfo.network,
- ipinfo.organization,
- ipinfo.static_ip_score,
- ipinfo.user_type,
- ipinfo.postal_code,
- CAST(ipinfo.latitude AS float) AS latitude,
- CAST(ipinfo.longitude AS float) AS longitude,
- ipinfo.accuracy_radius,
- ipinfo.ip,
- ipinfo.updated
- FROM thl_ipinformation AS ipinfo
- LEFT JOIN thl_geoname AS geo
- ON ipinfo.geoname_id = geo.geoname_id
- WHERE ipinfo.ip = %s
- """
- res = self.pg_config.execute_sql_query(query=query, params=[ip_address])
- assert len(res) == 1
- d = res[0]
- if d.get("geo_country_iso") and (d["geo_country_iso"] != d["country_iso"]):
- raise ValueError(
- f'mismatch between ipinfo country {d["country_iso"]} and geoname country {d["geo_country_iso"]}'
+ ips = set(ip_addresses)
+ d = {ip: self.grip_mmdb.lookup(ip) for ip in ips}
+ d = {
+ ip: GeoIPInformation(
+ ip=ip,
+ country_iso=v.country.country_iso.lower()
+ if v.country.country_iso
+ else None,
+ is_anonymous=v.anonymous.is_anonymous,
+ autonomous_system_number=v.asn.asn,
+ autonomous_system_organization=v.asn.network_operator,
+ access_type=v.asn.access_type,
)
+ for ip, v in d.items()
+ }
return d
-
- def get_mysql(self, ip_address: IPvAnyAddressStr):
- normalized_ip, lookup_prefix = normalize_ip(ip_address)
- d = self.get_mysql_raw(normalized_ip)
- g = GeoIPInformation.from_mysql(d)
- g.ip = ip_address
- g.lookup_prefix = lookup_prefix
- return g
-
- def recreate_cache(self, ip_address: IPvAnyAddressStr) -> GeoIPInformation:
- res = self.get_mysql(ip_address)
- self.set_cache(res)
- return res
-
- def get_mysql_multi(
- self,
- ips: Collection[IPvAnyAddressStr],
- ) -> dict[IPvAnyAddressStr, GeoIPInformation | None]:
-
- if len(ips) == 0:
- return {}
-
- with self.pg_config.make_connection() as conn, conn.cursor() as c:
- res = {}
- for chunk in chunked(ips, 500):
- inner = self.get_mysql_multi_chunk(
- c=c,
- ips=chunk,
- )
- res.update(inner)
- return res
-
- def get_mysql_multi_chunk(
- self,
- c: Cursor,
- ips: list[IPvAnyAddressStr],
- ) -> dict[IPvAnyAddressStr, GeoIPInformation | None]:
-
- assert len(ips) <= 500, "chunk me"
-
- # We must do it like this b/c we could have multiple /128 ips that normalize
- # to the same normalized /64 ip, and we don't want to "loose" them.
- ip_norm_lookup = {ip: normalize_ip(ip) for ip in ips}
- normalized_ips = {v[0] for v in ip_norm_lookup.values()}
- # also lookup exact matches (can remove this 28 days from 2025-11-15)
- normalized_ips.update(ips)
- # also lookup compressed form ... (remove me also)
- normalized_ips.update({self.compress_ip(ip) for ip in ips})
-
- c.execute(
- query="""
- SELECT
- geo.geoname_id,
- geo.continent_name,
- LOWER(geo.continent_code) AS continent_code,
- geo.country_name,
- LOWER(geo.country_iso) AS geo_country_iso,
- geo.subdivision_1_iso,
- geo.subdivision_1_name,
- geo.subdivision_2_iso,
- geo.subdivision_2_name,
- geo.city_name,
- geo.metro_code,
- geo.time_zone,
- geo.is_in_european_union,
- LOWER(ipinfo.country_iso) AS country_iso,
- ipinfo.registered_country_iso,
- ipinfo.is_anonymous,
- ipinfo.is_anonymous_vpn,
- ipinfo.is_hosting_provider,
- ipinfo.is_public_proxy,
- ipinfo.is_tor_exit_node,
- ipinfo.is_residential_proxy,
- ipinfo.autonomous_system_number,
- ipinfo.autonomous_system_organization,
- ipinfo.domain,
- ipinfo.isp,
- ipinfo.mobile_country_code,
- ipinfo.mobile_network_code,
- ipinfo.network,
- ipinfo.organization,
- ipinfo.static_ip_score,
- ipinfo.user_type,
- ipinfo.postal_code,
- CAST(ipinfo.latitude AS float) AS latitude,
- CAST(ipinfo.longitude AS float) AS longitude,
- ipinfo.accuracy_radius,
- ipinfo.ip,
- ipinfo.updated
- FROM thl_ipinformation AS ipinfo
- LEFT JOIN thl_geoname AS geo
- ON ipinfo.geoname_id = geo.geoname_id
- WHERE ipinfo.ip = ANY(%s)
- """,
- params=[list(normalized_ips)],
- )
-
- res = c.fetchall()
- for d in res:
- if d.get("geo_country_iso") and (d["geo_country_iso"] != d["country_iso"]):
- raise ValueError(
- f'mismatch between ipinfo country {d["country_iso"]} and geoname country {d["geo_country_iso"]}'
- )
-
- gs = [GeoIPInformation.from_mysql(i) for i in res]
- gs = {g.ip: g for g in gs}
- res2 = {}
-
- for ip, (normalized_ip, lookup_prefix) in ip_norm_lookup.items():
- if normalized_ip not in gs:
- # also can remove 28 days after 2025-11-15
- if ip in gs:
- res2[ip] = gs[ip].model_copy()
- continue
- res2[ip] = None
- continue
- g = gs[normalized_ip]
- g.ip = ip
- g.lookup_prefix = lookup_prefix
- res2[g.ip] = g.model_copy()
- return res2
diff --git a/generalresearch/managers/thl/payout.py b/generalresearch/managers/thl/payout.py
index 59dc2e9..4014cf8 100644
--- a/generalresearch/managers/thl/payout.py
+++ b/generalresearch/managers/thl/payout.py
@@ -34,10 +34,6 @@ from generalresearch.models.thl.payout import (
PayoutEvent,
UserPayoutEvent,
)
-from generalresearch.models.thl.wallet.cashout_method import (
- CashMailOrderData,
- CashoutRequestInfo,
-)
from generalresearch.models.thl.wallet.definitions import PayoutType
if TYPE_CHECKING:
@@ -105,225 +101,6 @@ class PayoutEventManager(PostgresManagerWithRedis):
conn.commit()
-class UserPayoutEventManager(PayoutEventManager):
- def get_by_uuid(self, pe_uuid: UUIDStr) -> UserPayoutEvent:
-
- res = self.pg_config.execute_sql_query(
- query="""
- SELECT ep.uuid,
- ep.debit_account_uuid,
- ep.cashout_method_uuid,
- ep.created, ep.amount, ep.status, ep.ext_ref_id, ep.payout_type,
- ep.request_data::jsonb,
- ep.order_data::jsonb,
- -- User Payout specific
- ac.name as description,
- la.reference_type as account_reference_type,
- la.reference_uuid as account_reference_uuid
- FROM event_payout AS ep
- LEFT JOIN accounting_cashoutmethod AS ac
- ON ep.cashout_method_uuid = ac.id
- LEFT JOIN ledger_account AS la
- ON la.uuid = ep.debit_account_uuid
- WHERE ep.uuid = %s
- """,
- params=[pe_uuid],
- )
-
- assert len(res) == 1, f"{pe_uuid} expected 1 result, got {len(res)}"
-
- d = res[0]
- pe = UserPayoutEvent.model_validate(d)
- if pe.order_data and pe.payout_type == PayoutType.CASH_IN_MAIL:
- pe.order_data = CashMailOrderData.model_validate(pe.order_data)
-
- return pe
-
- def get_payout_detail(self, pe_uuid: UUIDStr) -> CashoutRequestInfo:
- # This gets the payout event, and then extracts information for
- # the purposes of returning to the user.
- pe = self.get_by_uuid(pe_uuid=pe_uuid)
-
- transaction_info = {}
- order: dict[str, Any] = pe.order_data
- if pe.payout_type == PayoutType.TANGO and pe.status == PayoutStatus.COMPLETE:
- reward = order["reward"]
- if "credentialList" in reward:
- reward["credential_list"] = reward.pop("credentialList")
- if "redemptionInstructions" in reward:
- reward["redemption_instructions"] = reward.pop("redemptionInstructions")
- transaction_info = order["reward"]
- elif pe.payout_type == PayoutType.PAYPAL and pe.status == PayoutStatus.COMPLETE:
- info = {"transaction_id": order["transaction_id"]}
- transaction_info = info
- elif (
- pe.payout_type == PayoutType.CASH_IN_MAIL
- and pe.status == PayoutStatus.COMPLETE
- ):
- transaction_info = pe.order_data.model_dump(mode="json")
-
- return CashoutRequestInfo(
- id=pe_uuid,
- status=pe.status,
- description=pe.description,
- transaction_info=transaction_info,
- message="",
- )
-
- def filter_by(
- self,
- reference_uuid: str | None = None,
- debit_account_uuids: Collection[UUIDStr] | None = None,
- amount: int | None = None,
- created: datetime | None = None,
- created_after: datetime | None = None,
- product_ids: str | None = None,
- bp_user_ids: Collection[str] | None = None,
- cashout_method_uuids: Collection[UUIDStr] | None = None,
- cashout_types: Collection[PayoutType] | None = None,
- statuses: Collection[PayoutStatus] | None = None,
- ) -> list[UserPayoutEvent]:
- """Try to retrieve payout events by the product_id/user_uuid, amount,
- and optionally timestamp.
-
- WARNING: This is only on the "payout events" table and nothing to
- do with the Ledger itself. Therefore, the product_ids query
- doesn't return Brokerage Product Payouts (the ACH or Wire events
- to Suppliers) as part of the query.
-
- *** IT IS ONLY FOR USER PAYOUTS ***
-
- Note: what used to be in thl-grpcs "ListCashoutRequests" calling
- "list_cashout_requests" was merged into this.
- """
- args = []
- filters = []
-
- if reference_uuid:
- # This could be a product_id or a user_uuid
- filters.append("la.reference_uuid = %s")
- args.append(reference_uuid)
-
- if debit_account_uuids:
- # Or we could use the bp_wallet or user_wallet's account uuid
- # instead of looking up by the product/user
- filters.append("ep.debit_account_uuid = ANY(%s)")
- args.append(debit_account_uuids)
- if amount:
- filters.append("ep.amount = %s")
- args.append(amount)
- if created:
- filters.append("ep.created = %s")
- args.append(created.replace(tzinfo=None))
- if created_after:
- filters.append("ep.created >= %s")
- args.append(created_after.replace(tzinfo=None))
- if product_ids:
- filters.append("product_id = ANY(%s)")
- args.append(product_ids)
- if bp_user_ids:
- filters.append("product_user_id = ANY(%s)")
- args.append(bp_user_ids)
- if cashout_method_uuids:
- filters.append("cashout_method_uuid = ANY(%s)")
- args.append(cashout_method_uuids)
- if cashout_types:
- filters.append("payout_type = ANY(%s)")
- args.append([x.value for x in cashout_types])
- if statuses:
- filters.append("status = ANY(%s)")
- args.append([x.value for x in statuses])
-
- assert len(filters) > 0, "must pass at least 1 filter"
- filter_str = "WHERE " + " AND ".join(filters)
-
- res = self.pg_config.execute_sql_query(
- query=f"""
- SELECT
- ep.uuid, ep.debit_account_uuid,
- ep.created, ep.amount, ep.status,
- ep.ext_ref_id, ep.payout_type, ep.cashout_method_uuid,
- ep.order_data::jsonb,
- ep.request_data::jsonb,
- ac.name as description,
- la.reference_type as account_reference_type,
- la.reference_uuid as account_reference_uuid
- FROM event_payout AS ep
- LEFT JOIN accounting_cashoutmethod AS ac
- ON ep.cashout_method_uuid = ac.id
- LEFT JOIN ledger_account AS la
- ON la.uuid = ep.debit_account_uuid
- LEFT JOIN thl_user u
- ON la.reference_uuid = u.uuid
- {filter_str}
- """,
- params=args,
- )
-
- pes = []
- for d in res:
- pes.append(UserPayoutEvent.model_validate(d))
- return pes
-
- def create(
- self,
- debit_account_uuid: UUIDStr,
- cashout_method_uuid: UUIDStr,
- payout_type: PayoutType,
- amount: PositiveInt,
- # --- Optional: Default / Default Factory ---
- uuid: UUIDStr | None = None,
- status: PayoutStatus | None = None,
- created: AwareDatetimeISO | None = None,
- request_data: dict[str, Any] | None = None,
- # --- Optional: None ---
- account_reference_type: str | None = None,
- account_reference_uuid: UUIDStr | None = None,
- description: str | None = None,
- ext_ref_id: str | None = None,
- order_data: dict[str, Any] | CashMailOrderData | None = None,
- ) -> UserPayoutEvent:
-
- payout_event = UserPayoutEvent(
- uuid=uuid or uuid4().hex,
- debit_account_uuid=debit_account_uuid,
- account_reference_type=account_reference_type,
- account_reference_uuid=account_reference_uuid,
- cashout_method_uuid=cashout_method_uuid,
- description=description,
- created=created or datetime.now(tz=UTC),
- amount=amount,
- status=status or PayoutStatus.PENDING,
- ext_ref_id=ext_ref_id,
- payout_type=payout_type,
- request_data=request_data or {},
- order_data=order_data,
- )
- d = payout_event.model_dump_postgres()
-
- with self.pg_config.make_connection() as conn:
- with conn.cursor() as c:
- c.execute(
- query="""
- INSERT INTO event_payout (
- uuid, debit_account_uuid, created,
- cashout_method_uuid, amount, status,
- ext_ref_id, payout_type, order_data,
- request_data
- ) VALUES (
- %(uuid)s, %(debit_account_uuid)s, %(created)s,
- %(cashout_method_uuid)s, %(amount)s, %(status)s,
- %(ext_ref_id)s, %(payout_type)s, %(order_data)s,
- %(request_data)s
- );
- """,
- params=d,
- )
- assert c.rowcount == 1, f"expected 1 row inserted, got {c.rowcount}"
- conn.commit()
-
- return payout_event
-
class BrokerageProductPayoutEventManager(PayoutEventManager):
# This is what makes a PayoutEvent a Brokerage Product Payout
diff --git a/generalresearch/managers/thl/userhealth.py b/generalresearch/managers/thl/userhealth.py
index b986256..0df67aa 100644
--- a/generalresearch/managers/thl/userhealth.py
+++ b/generalresearch/managers/thl/userhealth.py
@@ -6,7 +6,6 @@ from datetime import UTC, datetime, timedelta
from itertools import zip_longest
from typing import TYPE_CHECKING, Any
-import faker
from pydantic import NonNegativeInt, PositiveInt
from generalresearch.decorators import LOG
@@ -14,16 +13,17 @@ from generalresearch.managers.base import (
PostgresManager,
PostgresManagerWithRedis,
)
-from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
from generalresearch.models.thl.user_iphistory import (
IPRecord,
UserIPHistory,
UserIPRecord,
)
+from generalresearch.models.thl.user_ref import UserRef
from generalresearch.models.thl.userhealth import AuditLog
if TYPE_CHECKING:
from generalresearch.managers.base import Permission
+ from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
from generalresearch.models.custom_types import IPvAnyAddressStr
from generalresearch.models.thl.product import Product
from generalresearch.models.thl.user import User
@@ -31,46 +31,18 @@ if TYPE_CHECKING:
from generalresearch.pg_helper import PostgresConfig
from generalresearch.redis_helper import RedisConfig
-fake = faker.Faker()
-
class UserIpHistoryManager(PostgresManagerWithRedis):
- def __init__(
- self,
- pg_config: PostgresConfig,
- redis_config: RedisConfig,
- permissions: Collection[Permission] | None = None,
- cache_prefix: str | None = None,
- ):
- super().__init__(
- pg_config=pg_config,
- redis_config=redis_config,
- permissions=permissions,
- cache_prefix=cache_prefix,
- )
- self.geoipinfo_manager = GeoIpInfoManager(
- pg_config=pg_config,
- redis_config=redis_config,
- cache_prefix=cache_prefix,
- )
-
def get_redis_key(self, user_id: int) -> str:
- return f"generalreserach:user-ip-history:{user_id}"
+ return f"generalreserach:user-ip-history-v2:{user_id}"
def get_user_ip_records_sql(self, user_id: int) -> list[UserIPRecord]:
# The IP metadata is ONLY for the 'ip', NOT for any forwarded ips.
# This might get called immediately after a write, so use the non-rr
res = self.pg_config.execute_sql_query(
query="""
- SELECT iph.ip, iph.created, iph.user_id,
- geo.subdivision_1_iso,
- ipinfo.country_iso,
- ipinfo.is_anonymous
+ SELECT iph.ip, iph.created
FROM userhealth_iphistory iph
- LEFT JOIN thl_ipinformation AS ipinfo
- ON iph.ip = ipinfo.ip
- LEFT JOIN thl_geoname AS geo
- ON ipinfo.geoname_id = geo.geoname_id
WHERE iph.user_id = %s
AND created > NOW() - INTERVAL '28 days'
ORDER BY iph.created DESC
@@ -95,76 +67,96 @@ class UserIpHistoryManager(PostgresManagerWithRedis):
value = iph.model_dump_json()
self.redis_client.set(self.get_redis_key(user_id), value, ex=3 * 24 * 3600)
- def recreate_user_ip_history_cache(self, user_id: int) -> None:
- self.delete_user_ip_history_cache(user_id=user_id)
- records = self.get_user_ip_records_sql(user_id=user_id)
+ def recreate_user_ip_history_cache(self, user: UserRef) -> None:
+ self.delete_user_ip_history_cache(user_id=user.user_id)
+ records = self.get_user_ip_records_sql(user_id=user.user_id)
# todo: we may get dns records from somewhere else here ...
- iph = UserIPHistory(user_id=user_id, ips=records)
- self.set_user_ip_history_cache(user_id=user_id, iph=iph)
+ iph = UserIPHistory(user=user, ips=records)
+ self.set_user_ip_history_cache(user_id=user.user_id, iph=iph)
+
+ def get_user_ip_history(
+ self, user: UserRef | User, geoip_info_manager: GeoIpInfoManager | None = None
+ ) -> UserIPHistory:
+ user = user if isinstance(user, UserRef) else user.to_user_ref()
- def get_user_ip_history(self, user_id: int) -> UserIPHistory:
- assert isinstance(user_id, int)
- iph = self.get_user_ip_history_cache(user_id=user_id)
+ iph = self.get_user_ip_history_cache(user_id=user.user_id)
if iph:
LOG.debug(f"get_user_ip_history got in cache: {iph.model_dump_json()}")
-
else:
LOG.debug("get_user_ip_history cache not found, using mysql")
- records = self.get_user_ip_records_sql(user_id=user_id)
+ records = self.get_user_ip_records_sql(user_id=user.user_id)
# todo: we may get dns records from somewhere else here ...
- iph = UserIPHistory(user_id=user_id, ips=records)
- self.set_user_ip_history_cache(user_id=user_id, iph=iph)
-
- iph.enrich_ips(pg_config=self.pg_config, redis_config=self.redis_config)
+ iph = UserIPHistory(user=user, ips=records)
+ self.set_user_ip_history_cache(user_id=user.user_id, iph=iph)
+ if geoip_info_manager:
+ iph.enrich_ips(geoip_info_manager=geoip_info_manager)
return iph
- def get_user_latest_ip(self, user: User, exclude_anon: bool = False) -> str | None:
- record = self.get_user_latest_ip_record(user=user, exclude_anon=exclude_anon)
- if record:
- return record.ip
- return None
-
def get_user_latest_ip_record(
- self, user: User, exclude_anon: bool = False
+ self,
+ user: UserRef | User,
+ exclude_anon: bool = False,
+ geoip_info_manager: GeoIpInfoManager | None = None,
) -> UserIPRecord | None:
- iphistory = self.get_user_ip_history(user_id=user.user_id)
-
- if iphistory.ips:
- if exclude_anon:
- return next(
- filter(
- lambda x: not x.information.is_anonymous,
- iphistory.ips[::-1],
- ),
- None,
- )
- else:
- return iphistory.ips[-1]
+ iphistory = self.get_user_ip_history(
+ user=user, geoip_info_manager=geoip_info_manager
+ )
+ if not iphistory or not iphistory.ips:
+ return None
+
+ # Note: UserIPHistory.ips is sorted by 'created DESC' !!!
+ # This logic was changed at some point? Or the py-utils
+ # get_user_latest_ip_record was changed. Please be careful here...
+ if exclude_anon:
+ assert geoip_info_manager is not None, "Must pass geoip_info_manager"
+ for ipr in iphistory.ips:
+ if not ipr.information.is_anonymous:
+ return ipr
+ else:
+ ipr = iphistory.ips[0]
+ return ipr
+ def get_user_latest_ip(
+ self,
+ user: UserRef | User,
+ exclude_anon: bool = False,
+ geoip_info_manager: GeoIpInfoManager | None = None,
+ ) -> str | None:
+ record = self.get_user_latest_ip_record(
+ user=user, exclude_anon=exclude_anon, geoip_info_manager=geoip_info_manager
+ )
+ if record:
+ return record.ip
return None
def get_user_latest_country(
- self, user: User, exclude_anon: bool = False
+ self,
+ user: UserRef | User,
+ geoip_info_manager: GeoIpInfoManager,
+ exclude_anon: bool = False,
) -> str | None:
"""Get the country the user is in, based off their latest ip."""
- ipr = self.get_user_latest_ip_record(user, exclude_anon=exclude_anon)
+ ipr = self.get_user_latest_ip_record(
+ user, geoip_info_manager=geoip_info_manager, exclude_anon=exclude_anon
+ )
# The ipr.information should exist, but it is possible the user has
# no IP history at all, so the record is None
return ipr.country_iso if ipr is not None else None
- def is_user_anonymous(self, user: User) -> bool | None:
+ def is_user_anonymous(
+ self, user: UserRef | User, geoip_info_manager: GeoIpInfoManager
+ ) -> bool | None:
# Get the user's latest ip. is it marked as anonymous?
- # Note: it is possible we only did a "basic" lookup of this IP so
- # we don't know if they are anonymous. Default to False
# Return None if the user has no IP history at all
- ipr = self.get_user_latest_ip_record(user)
+ ipr = self.get_user_latest_ip_record(
+ user, geoip_info_manager=geoip_info_manager
+ )
if ipr:
return ipr.is_anonymous if ipr.is_anonymous is not None else False
return None
class IPRecordManager(PostgresManagerWithRedis):
-
def __init__(
self,
pg_config: PostgresConfig,
@@ -187,7 +179,7 @@ class IPRecordManager(PostgresManagerWithRedis):
def create_unpack(
self,
- user_id: PositiveInt,
+ user: UserRef,
ip: IPvAnyAddressStr,
forwarded_ips: list[str],
) -> IPRecord:
@@ -196,11 +188,11 @@ class IPRecordManager(PostgresManagerWithRedis):
padded = list(forwarded_ips) + [None] * (6 - len(forwarded_ips))
- return self.create(user_id, ip, *padded)
+ return self.create(user, ip, *padded)
def create(
self,
- user_id: PositiveInt,
+ user: UserRef,
ip: IPvAnyAddressStr,
forwarded_ip1: IPvAnyAddressStr,
forwarded_ip2: IPvAnyAddressStr,
@@ -211,7 +203,7 @@ class IPRecordManager(PostgresManagerWithRedis):
) -> IPRecord:
data = {
- "user_id": user_id,
+ "user_id": user.user_id,
"ip": ipaddress.ip_address(ip).exploded,
"created": datetime.now(tz=UTC),
}
@@ -253,11 +245,12 @@ class IPRecordManager(PostgresManagerWithRedis):
""",
params=data,
)
- self.recreate_user_ip_history_cache(user_id=user_id)
+ self.recreate_user_ip_history_cache(user=user)
return IPRecord.from_mysql(data)
- def get_user_latest_ip_record(self, user: User) -> IPRecord | None:
+ def get_user_latest_ip_record(self, user: UserRef | User) -> IPRecord | None:
+ user = user if isinstance(user, UserRef) else user.to_user_ref()
res = self.filter_ip_records(user_ids=[user.user_id], limit=1)
if res:
return res[0]
@@ -317,14 +310,11 @@ class IPRecordManager(PostgresManagerWithRedis):
return [IPRecord.from_mysql(i) for i in res]
- def recreate_user_ip_history_cache(self, user_id: int):
- return self.user_ip_history_manager.recreate_user_ip_history_cache(
- user_id=user_id
- )
+ def recreate_user_ip_history_cache(self, user: UserRef):
+ return self.user_ip_history_manager.recreate_user_ip_history_cache(user=user)
class AuditLogManager(PostgresManager):
-
def create(
self,
user_id: PositiveInt,
diff --git a/generalresearch/managers/thl/wallet/__init__.py b/generalresearch/managers/thl/wallet/__init__.py
index 1f0c30b..006908b 100644
--- a/generalresearch/managers/thl/wallet/__init__.py
+++ b/generalresearch/managers/thl/wallet/__init__.py
@@ -1,6 +1,7 @@
from decimal import Decimal
from typing import TYPE_CHECKING, Any
+from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
from generalresearch.managers.thl.wallet.approve import (
approve_amt_cashout,
approve_paypal_order,
@@ -34,6 +35,7 @@ def manage_pending_cashout(
user_ip_history_manager: UserIpHistoryManager,
user_manager: UserManager,
ledger_manager: ThlLedgerManager,
+ geoip_info_manager:GeoIpInfoManager,
order_data: dict[str, Any] | CashMailOrderData | None = None,
tango_client: TangoClient | None = None,
) -> UserPayoutEvent:
@@ -63,7 +65,7 @@ def manage_pending_cashout(
"manage_pending_cashout called on user without managed wallet"
)
assert not user.blocked, "manage_pending_cashout: Blocked user"
- assert not user_ip_history_manager.is_user_anonymous(user), (
+ assert not user_ip_history_manager.is_user_anonymous(user,geoip_info_manager=geoip_info_manager), (
"manage_pending_cashout: Anonymous user"
)
diff --git a/generalresearch/managers/thl/wallet/cashout_tasks.py b/generalresearch/managers/thl/wallet/cashout_tasks.py
new file mode 100644
index 0000000..b28b04f
--- /dev/null
+++ b/generalresearch/managers/thl/wallet/cashout_tasks.py
@@ -0,0 +1,3 @@
+
+
+
diff --git a/generalresearch/managers/thl/wallet/tango.py b/generalresearch/managers/thl/wallet/tango.py
index aa780c3..fc6763b 100644
--- a/generalresearch/managers/thl/wallet/tango.py
+++ b/generalresearch/managers/thl/wallet/tango.py
@@ -1,7 +1,6 @@
from __future__ import annotations
from datetime import timedelta
-from decimal import Decimal
from threading import Lock
from typing import Any
@@ -10,6 +9,11 @@ from cachetools import TTLCache, cachedmethod
from generalresearch.currency import USDCent
from generalresearch.managers.thl.cashout_method import CashoutMethodManager
from generalresearch.managers.thl.tango_api import TangoClient
+from generalresearch.models.thl.wallet.cashout_method import (
+ TangoCashoutMethodData,
+ CashoutMethod,
+ TangoCashoutMethodRequestData,
+)
from generalresearch.models.thl.wallet.definitions import (
CURRENCY_FORMATTER,
Currency,
@@ -57,27 +61,26 @@ class TangoManager:
return self.tango_client.get_order(tango_order_id)
def make_request(
- self, amount_usd: Decimal, cashout_method: Any, external_ref_id: str
- ) -> dict[str, Any]:
+ self, amount: USDCent, cashout_method: CashoutMethod, payout_event_id: str
+ ) -> TangoCashoutMethodRequestData:
"""Build the data needed to place a Tango order."""
- assert type(amount_usd) is Decimal
+ assert type(amount) is USDCent
utid = cashout_method.data.utid
- amount: Decimal | float = amount_usd
currency = cashout_method.original_currency
- currency_code = getattr(currency, "value", currency)
- if currency_code and currency_code != "USD":
- amount = round(float(amount) / self.get_exchange_rates()[currency_code], 2)
- return {
- "accountIdentifier": self.tango_account_id,
- "customerIdentifier": self.tango_customer_id,
- "utid": utid,
- "amount": str(amount),
- "amount_usd": str(amount_usd),
- "campaign": "300large",
- "sendEmail": False,
- "externalRefID": external_ref_id,
- "description": self.get_name(utid),
- }
+ if currency and currency != Currency.USD:
+ amount = round(float(amount) / self.get_exchange_rates()[currency], 2)
+ return TangoCashoutMethodRequestData.model_validate(
+ {
+ "accountIdentifier": self.tango_account_id,
+ "customerIdentifier": self.tango_customer_id,
+ "utid": utid,
+ "amount": str(amount),
+ "campaign": "300large",
+ "sendEmail": False,
+ "externalRefID": payout_event_id,
+ "description": self.get_name(utid),
+ }
+ )
@cachedmethod(
cache=lambda self: self._name_cache,
diff --git a/generalresearch/managers/thl/wallet/user_payout.py b/generalresearch/managers/thl/wallet/user_payout.py
new file mode 100644
index 0000000..88505c0
--- /dev/null
+++ b/generalresearch/managers/thl/wallet/user_payout.py
@@ -0,0 +1,392 @@
+import uuid
+from collections.abc import Collection
+from datetime import UTC, datetime
+from typing import Any
+from uuid import uuid4
+
+import slack
+from pydantic import PositiveInt
+from redis import Redis
+
+from generalresearch.currency import USDCent
+from generalresearch.managers.thl.cashout_method import CashoutMethodManager
+from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
+from generalresearch.managers.thl.ledger_manager.thl_ledger import ThlLedgerManager
+from generalresearch.managers.thl.payout import PayoutEventManager
+from generalresearch.managers.thl.userhealth import UserIpHistoryManager
+from generalresearch.managers.thl.wallet.tango import TangoManager
+from generalresearch.models.custom_types import AwareDatetimeISO, UUIDStr
+from generalresearch.models.thl.definitions import PayoutStatus
+from generalresearch.models.thl.payout import UserPayoutEvent
+from generalresearch.models.thl.user import User
+from generalresearch.models.thl.wallet.cashout_method import (
+ CashMailOrderData,
+ CashoutMethod,
+ CashoutRequestInfo,
+ PaypalCashoutMethodRequestData,
+ TangoCashoutMethodRequestData,
+ CashMailCashoutMethodRequestData,
+)
+from generalresearch.models.thl.wallet.definitions import PayoutType
+
+
+class UserPayoutEventManager(PayoutEventManager):
+ def get_by_uuid(self, pe_uuid: UUIDStr) -> UserPayoutEvent:
+
+ res = self.pg_config.execute_sql_query(
+ query="""
+ SELECT ep.uuid,
+ ep.debit_account_uuid,
+ ep.cashout_method_uuid,
+ ep.created, ep.amount, ep.status, ep.ext_ref_id, ep.payout_type,
+ ep.request_data::jsonb,
+ ep.order_data::jsonb,
+ -- User Payout specific
+ ac.name as description,
+ la.reference_type as account_reference_type,
+ la.reference_uuid as account_reference_uuid
+ FROM event_payout AS ep
+ LEFT JOIN accounting_cashoutmethod AS ac
+ ON ep.cashout_method_uuid = ac.id
+ LEFT JOIN ledger_account AS la
+ ON la.uuid = ep.debit_account_uuid
+ WHERE ep.uuid = %s
+ """,
+ params=[pe_uuid],
+ )
+
+ assert len(res) == 1, f"{pe_uuid} expected 1 result, got {len(res)}"
+
+ d = res[0]
+ pe = UserPayoutEvent.model_validate(d)
+ if pe.order_data and pe.payout_type == PayoutType.CASH_IN_MAIL:
+ pe.order_data = CashMailOrderData.model_validate(pe.order_data)
+
+ return pe
+
+ def get_payout_detail(self, pe_uuid: UUIDStr) -> CashoutRequestInfo:
+ # This gets the payout event, and then extracts information for
+ # the purposes of returning to the user.
+ pe = self.get_by_uuid(pe_uuid=pe_uuid)
+
+ transaction_info = {}
+ order: dict[str, Any] = pe.order_data
+ if pe.payout_type == PayoutType.TANGO and pe.status == PayoutStatus.COMPLETE:
+ reward = order["reward"]
+ if "credentialList" in reward:
+ reward["credential_list"] = reward.pop("credentialList")
+ if "redemptionInstructions" in reward:
+ reward["redemption_instructions"] = reward.pop("redemptionInstructions")
+ transaction_info = order["reward"]
+ elif pe.payout_type == PayoutType.PAYPAL and pe.status == PayoutStatus.COMPLETE:
+ info = {"transaction_id": order["transaction_id"]}
+ transaction_info = info
+ elif (
+ pe.payout_type == PayoutType.CASH_IN_MAIL
+ and pe.status == PayoutStatus.COMPLETE
+ ):
+ transaction_info = pe.order_data.model_dump(mode="json")
+
+ return CashoutRequestInfo(
+ id=pe_uuid,
+ status=pe.status,
+ description=pe.description,
+ transaction_info=transaction_info,
+ message="",
+ )
+
+ def filter_by(
+ self,
+ reference_uuid: str | None = None,
+ debit_account_uuids: Collection[UUIDStr] | None = None,
+ amount: int | None = None,
+ created: datetime | None = None,
+ created_after: datetime | None = None,
+ product_ids: str | None = None,
+ bp_user_ids: Collection[str] | None = None,
+ cashout_method_uuids: Collection[UUIDStr] | None = None,
+ cashout_types: Collection[PayoutType] | None = None,
+ statuses: Collection[PayoutStatus] | None = None,
+ ) -> list[UserPayoutEvent]:
+ """Try to retrieve payout events by the product_id/user_uuid, amount,
+ and optionally timestamp.
+
+ WARNING: This is only on the "payout events" table and nothing to
+ do with the Ledger itself. Therefore, the product_ids query
+ doesn't return Brokerage Product Payouts (the ACH or Wire events
+ to Suppliers) as part of the query.
+
+ *** IT IS ONLY FOR USER PAYOUTS ***
+
+ Note: what used to be in thl-grpcs "ListCashoutRequests" calling
+ "list_cashout_requests" was merged into this.
+ """
+ args = []
+ filters = []
+
+ if reference_uuid:
+ # This could be a product_id or a user_uuid
+ filters.append("la.reference_uuid = %s")
+ args.append(reference_uuid)
+
+ if debit_account_uuids:
+ # Or we could use the bp_wallet or user_wallet's account uuid
+ # instead of looking up by the product/user
+ filters.append("ep.debit_account_uuid = ANY(%s)")
+ args.append(debit_account_uuids)
+ if amount:
+ filters.append("ep.amount = %s")
+ args.append(amount)
+ if created:
+ filters.append("ep.created = %s")
+ args.append(created.replace(tzinfo=None))
+ if created_after:
+ filters.append("ep.created >= %s")
+ args.append(created_after.replace(tzinfo=None))
+ if product_ids:
+ filters.append("product_id = ANY(%s)")
+ args.append(product_ids)
+ if bp_user_ids:
+ filters.append("product_user_id = ANY(%s)")
+ args.append(bp_user_ids)
+ if cashout_method_uuids:
+ filters.append("cashout_method_uuid = ANY(%s)")
+ args.append(cashout_method_uuids)
+ if cashout_types:
+ filters.append("payout_type = ANY(%s)")
+ args.append([x.value for x in cashout_types])
+ if statuses:
+ filters.append("status = ANY(%s)")
+ args.append([x.value for x in statuses])
+
+ assert len(filters) > 0, "must pass at least 1 filter"
+ filter_str = "WHERE " + " AND ".join(filters)
+
+ res = self.pg_config.execute_sql_query(
+ query=f"""
+ SELECT
+ ep.uuid, ep.debit_account_uuid,
+ ep.created, ep.amount, ep.status,
+ ep.ext_ref_id, ep.payout_type, ep.cashout_method_uuid,
+ ep.order_data::jsonb,
+ ep.request_data::jsonb,
+ ac.name as description,
+ la.reference_type as account_reference_type,
+ la.reference_uuid as account_reference_uuid
+ FROM event_payout AS ep
+ LEFT JOIN accounting_cashoutmethod AS ac
+ ON ep.cashout_method_uuid = ac.id
+ LEFT JOIN ledger_account AS la
+ ON la.uuid = ep.debit_account_uuid
+ LEFT JOIN thl_user u
+ ON la.reference_uuid = u.uuid
+ {filter_str}
+ """,
+ params=args,
+ )
+
+ pes = []
+ for d in res:
+ pes.append(UserPayoutEvent.model_validate(d))
+ return pes
+
+ def create(
+ self,
+ debit_account_uuid: UUIDStr,
+ cashout_method_uuid: UUIDStr,
+ payout_type: PayoutType,
+ amount: PositiveInt,
+ # --- Optional: Default / Default Factory ---
+ uuid: UUIDStr | None = None,
+ status: PayoutStatus | None = None,
+ created: AwareDatetimeISO | None = None,
+ request_data: dict[str, Any] | None = None,
+ # --- Optional: None ---
+ account_reference_type: str | None = None,
+ account_reference_uuid: UUIDStr | None = None,
+ description: str | None = None,
+ ext_ref_id: str | None = None,
+ order_data: dict[str, Any] | CashMailOrderData | None = None,
+ ) -> UserPayoutEvent:
+
+ payout_event = UserPayoutEvent(
+ uuid=uuid or uuid4().hex,
+ debit_account_uuid=debit_account_uuid,
+ account_reference_type=account_reference_type,
+ account_reference_uuid=account_reference_uuid,
+ cashout_method_uuid=cashout_method_uuid,
+ description=description,
+ created=created or datetime.now(tz=UTC),
+ amount=amount,
+ status=status or PayoutStatus.PENDING,
+ ext_ref_id=ext_ref_id,
+ payout_type=payout_type,
+ request_data=request_data or {},
+ order_data=order_data,
+ )
+ d = payout_event.model_dump_postgres()
+
+ with self.pg_config.make_connection() as conn:
+ with conn.cursor() as c:
+ c.execute(
+ query="""
+ INSERT INTO event_payout (
+ uuid, debit_account_uuid, created,
+ cashout_method_uuid, amount, status,
+ ext_ref_id, payout_type, order_data,
+ request_data
+ ) VALUES (
+ %(uuid)s, %(debit_account_uuid)s, %(created)s,
+ %(cashout_method_uuid)s, %(amount)s, %(status)s,
+ %(ext_ref_id)s, %(payout_type)s, %(order_data)s,
+ %(request_data)s
+ );
+ """,
+ params=d,
+ )
+ assert c.rowcount == 1, f"expected 1 row inserted, got {c.rowcount}"
+ conn.commit()
+
+ return payout_event
+
+ def user_request_redeem(
+ self,
+ user: User,
+ country_iso: str,
+ cashout_method_id: str,
+ amount: USDCent,
+ tango_manager: TangoManager,
+ cashout_method_manager: CashoutMethodManager,
+ ledger_manager: ThlLedgerManager,
+ user_ip_history_manager: UserIpHistoryManager,
+ geoip_info_manager: GeoIpInfoManager,
+ redis_client: Redis,
+ slack_client: slack.WebClient | None = None,
+ ) -> UserPayoutEvent:
+ """
+ A user has requested to be paid from their wallet balance.
+ :param user: User
+ :param cashout_method_id: the cashout method to use
+ :param amount_usd: In decimal USD.
+ """
+ now = datetime.now(tz=UTC)
+ user.prefetch_product(pg_config=self.pg_config)
+ usd_exchange_rates = tango_manager.get_exchange_rates()
+ cashout_methods = cashout_method_manager.get_user_cashout_methods(
+ user, country_iso=country_iso, usd_exchange_rate=usd_exchange_rates
+ )
+ assert cashout_method_id in cashout_methods, (
+ f"cashout method '{cashout_method_id}' not found"
+ )
+ cashout_method = cashout_methods[cashout_method_id]
+
+ # Make sure the amount requested is allowed for this cashout method
+ try:
+ cashout_method.validate_requested_amount(amount=amount)
+ except ValueError as e:
+ # Outer function is expecting an assertion error.
+ raise AssertionError(str(e))
+
+ """
+ These checks are run when a user requests any kind of payout:
+ - user wallet enabled on BP
+ - is user blocked
+ - is user anonymous
+ - does the user have a redeemable_balance <= amount requested (except for
+ an AMT assignment request)
+ """
+ assert type(amount) is USDCent
+ user.prefetch_product(pg_config=self.pg_config)
+ assert user.product.user_wallet_enabled, (
+ "SubmitUserCashout called on user without managed wallet"
+ )
+ assert not user.blocked, "Blocked user requesting redemption"
+ assert 0 < amount <= 250_00, (
+ f"Amount must be between 0 and $250.00. Got {amount.to_usd_str()}"
+ )
+
+ product = user.product
+ banned_countries = user.product.user_health_config.banned_countries
+
+ assert not user_ip_history_manager.is_user_anonymous(
+ user, geoip_info_manager=geoip_info_manager
+ ), "Anonymous user requesting redemption"
+ ipr = user_ip_history_manager.get_user_latest_ip_record(
+ user, geoip_info_manager=geoip_info_manager
+ )
+ if ipr is not None and ipr.country_iso in banned_countries:
+ raise AssertionError("Banned country requesting redemption")
+
+ wallet_balance = ledger_manager.get_user_wallet_balance(user)
+ if product.user_wallet_config.balance_type == "wallet_balance":
+ redeemable_amount = wallet_balance
+ elif product.user_wallet_config.balance_type == "redeemable_balance":
+ redeemable_amount = ledger_manager.get_user_redeemable_wallet_balance(user, wallet_balance)
+ else:
+ raise ValueError(f"unexpected balance_type={product.user_wallet_config.balance_type}")
+
+ assert amount <= redeemable_amount, (
+ f"User requesting more than their redeemable balance ({amount} > {redeemable_amount}}"
+ )
+
+ # Simple dedupe mechanism. Don't allow more than 1 per user_id per minute per cashout_method.
+ flag_just_set = bool(
+ redis_client.set(
+ f"user_request_redeem:{user.user_id}:{cashout_method}",
+ 1,
+ nx=True,
+ ex=60,
+ )
+ )
+ assert flag_just_set, "User requesting more than 1 per min per cashout method"
+
+ payout_type = cashout_method.type
+
+ if slack_client:
+ slack_client.chat_postMessage(
+ channel="cashouts",
+ text=f"{user.product.name} {cashout_method.type} "
+ f"{user.user_id} {amount.to_usd_str()}",
+ )
+
+ pe_uuid = uuid.uuid4().hex
+
+ if payout_type == PayoutType.TANGO:
+ request_data: TangoCashoutMethodRequestData = tango_manager.make_request(
+ amount, cashout_method, pe_uuid
+ )
+ elif payout_type == PayoutType.PAYPAL:
+ request_data: PaypalCashoutMethodRequestData = make_request_paypal(cashout_method)
+ elif payout_type == PayoutType.CASH_IN_MAIL:
+ request_data: CashMailCashoutMethodRequestData = CashMailCashoutMethodRequestData.model_validate(cashout_method.data.model_dump())
+ else:
+ raise ValueError(f"unknown {payout_type=}")
+
+ account = ledger_manager.get_account_or_create_user_wallet(user)
+
+ pe = self.create(
+ uuid=pe_uuid,
+ debit_account_uuid=account.uuid,
+ cashout_method_uuid=cashout_method_id,
+ amount=amount,
+ created=now,
+ payout_type=payout_type,
+ request_data=request_data.model_dump(mode="json"),
+ )
+
+ ledger_manager.create_tx_user_payout_request(user, payout_event=pe, created=now)
+ return pe
+
+
+
+
+
+def make_request_paypal(
+ cashout_method: CashoutMethod,
+) -> PaypalCashoutMethodRequestData:
+ return PaypalCashoutMethodRequestData.model_validate(
+ {
+ "email": cashout_method.data.email,
+ "interface": "api",
+ }
+ )
diff --git a/generalresearch/models/thl/ipinfo.py b/generalresearch/models/thl/ipinfo.py
index c4ad50c..be9ac14 100644
--- a/generalresearch/models/thl/ipinfo.py
+++ b/generalresearch/models/thl/ipinfo.py
@@ -2,16 +2,17 @@ from __future__ import annotations
import ipaddress
from datetime import UTC, datetime
-from typing import TYPE_CHECKING, Any, Literal, Self
+from typing import Any, Literal, Self
from faker import Faker
+from grip_client import AccessType
from pydantic import (
BaseModel,
ConfigDict,
Field,
PositiveInt,
- PrivateAttr,
field_validator,
+ IPvAnyAddress,
)
from generalresearch.models.custom_types import (
@@ -19,17 +20,13 @@ from generalresearch.models.custom_types import (
CountryISOLike,
IPvAnyAddressStr,
)
-from generalresearch.models.thl.maxmind.definitions import UserType
-
-if TYPE_CHECKING:
- from generalresearch.managers.thl.ipinfo import IPGeonameManager
fake = Faker()
PrefixLength = Literal["/128", "/64", "/32"]
-def normalize_ip(ip: IPvAnyAddressStr) -> tuple[str, PrefixLength]:
+def normalize_ip(ip: str | IPvAnyAddress) -> tuple[str, PrefixLength]:
"""
Normalize an IP address for MySQL storage.
@@ -41,7 +38,7 @@ def normalize_ip(ip: IPvAnyAddressStr) -> tuple[str, PrefixLength]:
"""
addr = ipaddress.ip_address(ip)
if addr.version == 4:
- return ip, "/32"
+ return addr.exploded, "/32"
net64 = ipaddress.IPv6Network((addr, 64), strict=False)
return net64.network_address.exploded, "/64"
@@ -126,138 +123,34 @@ class IPGeoname(BaseModel):
class IPInformation(BaseModel):
- ip: IPvAnyAddressStr = Field()
- # This doesn't get stored in mysql/redis, b/c we only look up by the normalized ip
- lookup_prefix: PrefixLength | None = Field(default=None, exclude=True)
+ """
+ Fields we'll always pull from GRIP's mmdb files at minimum
+ """
- geoname_id: PositiveInt | None = Field(default=None)
+ ip: IPvAnyAddressStr = Field()
- country_iso: CountryISOLike = Field(
+ country_iso: CountryISOLike | None = Field(
description="The ISO code of the country associated with the IP address.",
examples=[fake.country_code().lower()],
)
- registered_country_iso: CountryISOLike | None = Field(
- default=None,
- description="The ISO code of the country where the IP address is registered.",
- examples=[fake.country_code().lower()],
- )
is_anonymous: bool | None = Field(
default=None,
description="Indicates whether the IP address is associated with an "
"anonymous source (e.g., VPN, proxy).",
examples=[False],
)
- is_anonymous_vpn: bool | None = Field(default=None)
- is_hosting_provider: bool | None = Field(default=None)
- is_public_proxy: bool | None = Field(default=None)
- is_tor_exit_node: bool | None = Field(default=None)
- is_residential_proxy: bool | None = Field(default=None)
autonomous_system_number: PositiveInt | None = Field(default=None)
autonomous_system_organization: str | None = Field(default=None, max_length=255)
- domain: str | None = Field(default=None, max_length=255)
- isp: str | None = Field(
- default=None,
- description="The Internet Service Provider associated with the IP address.",
- examples=["Comcast"],
- )
-
- mobile_country_code: str | None = Field(default=None, max_length=3)
- mobile_network_code: str | None = Field(default=None, max_length=3)
-
- network: str | None = Field(default=None, max_length=56)
- organization: str | None = Field(default=None, max_length=255)
-
- static_ip_score: float | None = Field(
- default=None,
- description="A score indicating the likelihood that the IP address is static.",
- )
- user_type: UserType | None = Field(
+ access_type: AccessType | None = Field(
default=None,
description="The type of user associated with the IP address "
"(e.g., 'residential', 'business').",
- examples=[UserType.RESIDENTIAL],
- )
- postal_code: str | None = Field(
- default=None,
- description="The postal code associated with the IP address.",
- examples=[fake.postcode()],
- )
-
- latitude: float | None = Field(
- description="The latitude coordinate of the IP address location.",
- default=None,
- examples=[float(fake.latitude())],
- )
- longitude: float | None = Field(
- description="The longitude coordinate of the IP address location.",
- default=None,
- examples=[float(fake.longitude())],
- )
-
- accuracy_radius: int | None = Field(
- default=None,
- description="The approximate radius of accuracy for the latitude "
- "and longitude, in kilometers.",
- examples=[fake.random_int(min=25, max=250)],
- )
-
- updated: AwareDatetimeISO = Field(
- default_factory=lambda: datetime.now(tz=UTC),
+ examples=[AccessType.RESIDENTIAL],
)
- _geoname: IPGeoname | None = PrivateAttr(default=None)
-
- @field_validator("country_iso", "registered_country_iso", mode="before")
- def make_lower(cls, value: str | None) -> str | None:
- if value is not None:
- return value.lower()
-
- return value
-
- @property
- def basic(self) -> bool:
- # This could be almost any field, but we're checking here if GRIP
- # was run on this record. If not, then most of the optional
- # fields will be None
- return self.is_anonymous is None
-
- @property
- def geoname(self) -> IPGeoname | None:
- return self._geoname or None
-
- def normalize_ip(self):
- normalized_ip, lookup_prefix = normalize_ip(self.ip)
- self.ip = normalized_ip
- self.lookup_prefix = lookup_prefix
-
- # --- prefetch_* ---
- def prefetch_geoname(
- self,
- ip_gm: IPGeonameManager,
- ) -> None:
- if self.geoname_id is None:
- raise ValueError("Must provide geoname_id")
-
- # from generalresearch.managers.thl.ipinfo import IPGeonameManager
- # ip_gm = IPGeonameManager(pg_config=pg_config)
-
- self._geoname = ip_gm.get_by_id(geoname_id=self.geoname_id)
-
- # --- ORM ---
- def model_dump_mysql(self):
- d = self.model_dump(mode="json")
- d["updated"] = self.updated
- return d
-
- @classmethod
- def from_mysql(cls, d: dict[str, Any]) -> Self:
- d["updated"] = d["updated"].replace(tzinfo=UTC)
-
- return cls.model_validate(d)
-
-class GeoIPInformation(IPInformation, IPGeoname):
+class GeoIPInformation(IPInformation):
model_config = ConfigDict(extra="ignore")
diff --git a/generalresearch/models/thl/maxmind/__init__.py b/generalresearch/models/thl/maxmind/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/generalresearch/models/thl/maxmind/__init__.py
+++ /dev/null
diff --git a/generalresearch/models/thl/maxmind/definitions.py b/generalresearch/models/thl/maxmind/definitions.py
deleted file mode 100644
index 01431c7..0000000
--- a/generalresearch/models/thl/maxmind/definitions.py
+++ /dev/null
@@ -1,22 +0,0 @@
-from enum import Enum
-
-from generalresearch.utils.enum import ReprEnumMeta
-
-
-class UserType(Enum, metaclass=ReprEnumMeta):
- # https://support.maxmind.com/hc/en-us/articles/4408430082971-IP-Trait-Risk-Data#h_01FN6V8JMQMWZGWNPPAW77ZPY4
- BUSINESS = "business"
- CAFE = "cafe"
- CELLULAR = "cellular"
- COLLEGE = "college"
- CDN = "content_delivery_network"
- CPN = "consumer_privacy_network"
- GOVERNMENT = "government"
- HOSTING = "hosting"
- LIBRARY = "library"
- MILITARY = "military"
- RESIDENTIAL = "residential"
- ROUTER = "router"
- SCHOOL = "school"
- SEARCH_ENGINE = "search_engine_spider"
- TRAVELER = "traveler"
diff --git a/generalresearch/models/thl/user_iphistory.py b/generalresearch/models/thl/user_iphistory.py
index a7eadf4..0e09de0 100644
--- a/generalresearch/models/thl/user_iphistory.py
+++ b/generalresearch/models/thl/user_iphistory.py
@@ -4,7 +4,6 @@ import ipaddress
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Self
-from faker import Faker
from pydantic import (
BaseModel,
ConfigDict,
@@ -19,15 +18,12 @@ from generalresearch.models.custom_types import (
IPvAnyAddressStr,
)
from generalresearch.models.thl.ipinfo import GeoIPInformation, normalize_ip
-from generalresearch.models.thl.user import User
+from generalresearch.models.thl.user_ref import UserRef
if TYPE_CHECKING:
from grip_client.enums import AccessType
- from generalresearch.pg_helper import PostgresConfig
- from generalresearch.redis_helper import RedisConfig
-
-fake = Faker()
+ from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
class UserIPRecord(BaseModel):
@@ -43,30 +39,12 @@ class UserIPRecord(BaseModel):
@property
def is_anonymous(self) -> bool:
- # default to False even if insights is not looked up
- return (
- self.information.is_anonymous
- if self.information
- and self.information.basic is False
- and self.information.is_anonymous is not None
- else False
- )
-
- @property
- def user_type(self) -> AccessType | None:
- return self.information.user_type if self.information else None
+ # Default to false if information is not looked up
+ return (self.information.is_anonymous if self.information else False) or False
@property
def access_type(self) -> AccessType | None:
- return self.information.user_type if self.information else None
-
- @property
- def subdivision_1_iso(self) -> str | None:
- return self.information.subdivision_1_iso if self.information else None
-
- @property
- def subdivision_2_iso(self) -> str | None:
- return self.information.subdivision_2_iso if self.information else None
+ return self.information.access_type if self.information else None
class IPRecord(BaseModel):
@@ -93,27 +71,6 @@ class IPRecord(BaseModel):
) -> bool:
return not (ip == self.ip and forwarded_ips == self.forwarded_ips)
- # --- prefetch_* ---
- def prefetch_ipinfo(
- self,
- pg_config: PostgresConfig,
- redis_config: RedisConfig,
- include_forwarded: bool = True,
- ) -> None:
- from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
-
- m = GeoIpInfoManager(pg_config=pg_config, redis_config=redis_config)
-
- if include_forwarded:
- ips = {self.ip}
- ips.update(set(self.forwarded_ips))
- res = m.get_multi(ips)
- self.information = res.get(self.ip)
- for x in self.forwarded_ip_records:
- x.information = res.get(x.ip)
- else:
- self.information = m.get(ip_address=self.ip)
-
# --- ORM ---
@classmethod
def from_mysql(cls, d: dict) -> Self:
@@ -146,7 +103,7 @@ class IPRecord(BaseModel):
class UserIPHistory(BaseModel):
model_config = ConfigDict(validate_assignment=True)
- user_id: PositiveInt = Field()
+ user: UserRef = Field()
# In thl-gprc, we run "audit_ip_history()", and so a user should
# get blocked after 100 IP switches or 30 unique IPs
@@ -165,9 +122,6 @@ class UserIPHistory(BaseModel):
default=None, description="These are any IP addresses that came in "
)
- # -- prefetch_ fields
- user: User | None = Field(default=None)
-
@field_validator("ips", mode="after")
@classmethod
def ips_timestamp(cls, ips):
@@ -180,30 +134,11 @@ class UserIPHistory(BaseModel):
reverse=True,
)
- def prefetch_user(
- self,
- pg_config: PostgresConfig,
- redis_config: RedisConfig,
- pg_config_rr: PostgresConfig,
- ) -> None:
- from generalresearch.managers.thl.user_manager.user_manager import (
- UserManager,
- )
-
- um = UserManager(
- pg_config=pg_config,
- pg_config_rr=pg_config_rr,
- redis=redis_config.dsn,
- )
- self.user = um.get_user(user_id=self.user_id)
-
- def enrich_ips(self, pg_config: PostgresConfig, redis_config: RedisConfig) -> None:
- from generalresearch.managers.thl.ipinfo import GeoIpInfoManager
-
- m = GeoIpInfoManager(pg_config=pg_config, redis_config=redis_config)
-
+ def enrich_ips(self, geoip_info_manager: GeoIpInfoManager) -> None:
+ if not self.ips:
+ return
ip_addresses = {x.ip for x in self.ips if x.information is None}
- res = m.get_multi(ip_addresses=ip_addresses)
+ res = geoip_info_manager.get_multi(ip_addresses=ip_addresses)
for x in self.ips:
if res.get(x.ip):
x.information = res[x.ip]
diff --git a/generalresearch/models/thl/wallet/cashout_method.py b/generalresearch/models/thl/wallet/cashout_method.py
index 1e96818..d9bc34c 100644
--- a/generalresearch/models/thl/wallet/cashout_method.py
+++ b/generalresearch/models/thl/wallet/cashout_method.py
@@ -214,6 +214,8 @@ class CashMailCashoutMethodData(BaseModel):
description="Delivery address where payment should be sent"
)
+class CashMailCashoutMethodRequestData(CashMailCashoutMethodData):
+ pass
class PaypalCashoutMethodData(BaseModel):
type: Literal[PayoutType.PAYPAL] = Field(default=PayoutType.PAYPAL)
@@ -223,6 +225,22 @@ class PaypalCashoutMethodData(BaseModel):
examples=["test@example.com"],
)
+class PaypalCashoutMethodRequestData(BaseModel):
+ email: EmailStr = Field(
+ description="Email address of the paypal user",
+ examples=["test@example.com"],
+ )
+ interface: Literal["api"] = Field(default="api")
+
+class TangoCashoutMethodRequestData(BaseModel):
+ accountIdentifier: str = Field()
+ customerIdentifier: str = Field()
+ utid: str = Field(description="tango utid")
+ amount: USDCent = Field(description="Amount to be paid out")
+ campaign: Literal["300large"]
+ sendEmail: bool = Field(default=False)
+ externalRefID: str = Field(description="External Ref ID")
+ description: str = Field()
class TangoCashoutMethodData(BaseModel):
type: Literal[PayoutType.TANGO] = Field(default=PayoutType.TANGO)