Skip to content

Commit aaeada9

Browse files
kristoisbergkristoisberg
authored
feat: add Threadless email and username modules (#647)
1 parent 13ede2b commit aaeada9

3 files changed

Lines changed: 377 additions & 0 deletions

File tree

‎user_scanner/email_scan/creator/threadless.py‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
"""Threadless email registration check via the public Artist Shops signup form."""
2+
3+
import re
4+
5+
from curl_cffi.requests.exceptions import RequestException
6+
7+
from user_scanner.core.impersonate import impersonate_request_async
8+
from user_scanner.core.result import Result
9+
10+
SIGNUP_URL = "https://profile.threadless.com/artist/shops/"
11+
CSRF_RE = re.compile(r"name=['\"]csrfmiddlewaretoken['\"]\s+value=['\"]([^'\"]+)")
12+
13+
14+
async def validate_threadless(email: str) -> Result:
15+
"""Check Threadless without creating an account or sending email."""
16+
show_url = "https://www.threadless.com"
17+
18+
try:
19+
page = await impersonate_request_async(SIGNUP_URL, allow_redirects=True)
20+
token = CSRF_RE.search(page.text)
21+
if page.status_code != 200 or not token:
22+
return Result.error(
23+
f"Could not read Threadless signup form (HTTP {page.status_code})",
24+
url=show_url,
25+
)
26+
27+
response = await impersonate_request_async(
28+
SIGNUP_URL,
29+
"POST",
30+
headers={"x-requested-with": "XMLHttpRequest"},
31+
data={
32+
"validate": "true",
33+
"email": email,
34+
"csrfmiddlewaretoken": token.group(1),
35+
},
36+
)
37+
if response.status_code != 200:
38+
return Result.error(
39+
f"Unexpected Threadless response: HTTP {response.status_code}",
40+
url=show_url,
41+
)
42+
43+
except RequestException as exc:
44+
return Result.error(exc, url=show_url)
45+
46+
try:
47+
data = response.json()
48+
except ValueError:
49+
return Result.error("Invalid Threadless response", url=show_url)
50+
51+
if data.get("is_valid") is False:
52+
return Result.taken(url=show_url)
53+
if data.get("is_valid") is True:
54+
return Result.available(url=show_url)
55+
return Result.error("Unexpected Threadless response", url=show_url)

‎user_scanner/user_scan/creator/threadless.py‎

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
import html
2+
import re
3+
from urllib.parse import quote, urljoin
4+
5+
from curl_cffi.requests.exceptions import RequestException
6+
7+
from user_scanner.core.impersonate import impersonate_request, impersonate_validate
8+
from user_scanner.core.result import Result
9+
10+
SIGNUP_URL = "https://profile.threadless.com/artist/shops/"
11+
CSRF_RE = re.compile(r"name=['\"]csrfmiddlewaretoken['\"]\s+value=['\"]([^'\"]+)")
12+
13+
14+
def validate_threadless(user: str) -> Result:
15+
show_url = f"https://www.threadless.com/@{user}"
16+
17+
def process(page):
18+
token = CSRF_RE.search(page.text)
19+
if page.status_code != 200 or not token:
20+
return Result.error(
21+
f"Could not read Threadless signup form (HTTP {page.status_code})"
22+
)
23+
24+
response = impersonate_request(
25+
SIGNUP_URL,
26+
"POST",
27+
headers={"x-requested-with": "XMLHttpRequest"},
28+
data={
29+
"validate": "true",
30+
"username": user,
31+
"csrfmiddlewaretoken": token.group(1),
32+
},
33+
)
34+
if response.status_code != 200:
35+
return Result.error(
36+
f"Unexpected Threadless response: HTTP {response.status_code}"
37+
)
38+
39+
match response.json():
40+
case {"is_valid": False}:
41+
extra, media = _profile(user)
42+
return Result.taken(extra=extra, media=media)
43+
case {"is_valid": True}:
44+
return Result.available()
45+
case _:
46+
return Result.error("Unexpected Threadless response")
47+
48+
return impersonate_validate(
49+
SIGNUP_URL,
50+
process,
51+
show_url=show_url,
52+
allow_redirects=True,
53+
)
54+
55+
56+
def _profile(user: str) -> tuple[dict, dict]:
57+
try:
58+
response = impersonate_request(
59+
f"https://www.threadless.com/@{quote(user, safe='')}",
60+
allow_redirects=True,
61+
)
62+
except RequestException:
63+
return {}, {}
64+
65+
if response.status_code != 200:
66+
return {}, {}
67+
68+
document = response.text
69+
username = _text(r'var userName = "([^"]+)"', document)
70+
if not username or username.casefold() != user.casefold():
71+
return {}, {}
72+
73+
shop = _text(r'<div class="artist-shop">(.*?)</div>', document) or ""
74+
shop_url = _text(r'<a href="([^"]+)" title="Visit my Artist Shop"', shop)
75+
links = re.findall(
76+
r'<a href="(https?://[^"]+)" target="_blank" title="[^"]+" rel="nofollow"',
77+
document,
78+
)
79+
website = _text(
80+
r'<div class="website">(?:(?!</div>).)*?<a href="([^"]+)"', document
81+
)
82+
shop_preview = _text(r'<img src="([^"]+)"', shop)
83+
return (
84+
{
85+
"id": _text(r"var profileId = (\d+);", document),
86+
"name": _text(r'<span class="name">\s*([^<]+)', document),
87+
"bio": _text(r'<div class="cover-statement">\s*<p>(.*?)</p>', document),
88+
"location": _text(
89+
r'<div class="location">\s*<h3>Location</h3>\s*([^<]+)', document
90+
),
91+
"following": _text(r'/following"[^>]*>\s*<span>(\d+)</span>', document),
92+
"followers": _text(r'/followers"[^>]*>\s*<span>(\d+)</span>', document),
93+
"member_since": _text(r"Member since ([^<]+)", document),
94+
"artist_shop": urljoin("https://www.threadless.com", shop_url)
95+
if shop_url
96+
else None,
97+
"artist_shop_slug": _text(r'var artistSlug = "([^"]+)";', document),
98+
"artist_shop_name": _text(
99+
r'title="Visit my Artist Shop"[^>]*>\s*([^<]+)</a>',
100+
shop,
101+
),
102+
"website": website,
103+
"links": ", ".join(map(html.unescape, links)),
104+
"threads_started": _number(r"([\d,]+) threads started", document),
105+
"designs_submitted": _number(r"([\d,]+) designs submitted", document),
106+
"designs_scored": _number(r"([\d,]+) designs scored", document),
107+
"avg_score_given": _text(r"Avg Score Given:\s*([\d.]+)", document),
108+
"is_printed_artist": _flag("isPrintedArtist", document),
109+
"is_submitted_artist": _flag("isSubmittedArtist", document),
110+
"is_shop_owner": _flag("isShopOwner", document),
111+
"facebook_connected": _flag("fbConnected", document),
112+
"is_hifiver": _flag("isHifiver", document),
113+
"is_shop_published": _flag("isShopPublished", document),
114+
"open_submissions": _flag("openSubs", document),
115+
"is_alumni": "is-alumni" in document,
116+
},
117+
{
118+
"avatar": _text(r'<meta property="og:image" content="([^"]+)"', document),
119+
"cover": _text(
120+
r'<header id="header" style="background-image: url\([\'\"]([^\'\"]+)',
121+
document,
122+
),
123+
"artist_shop_preview": shop_preview,
124+
},
125+
)
126+
127+
128+
def _text(pattern: str, document: str) -> str | None:
129+
match = re.search(pattern, document, re.DOTALL)
130+
return html.unescape(match.group(1)).strip() if match else None
131+
132+
133+
def _number(pattern: str, document: str) -> int | None:
134+
value = _text(pattern, document)
135+
return int(value.replace(",", "")) if value else None
136+
137+
138+
def _flag(name: str, document: str) -> bool | None:
139+
value = _text(rf"var {name} = (true|false);", document)
140+
return value == "true" if value else None

‎user_scanner/user_scan/creator/threadless_shop.py‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
import html
2+
import re
3+
4+
from curl_cffi.requests.exceptions import RequestException
5+
6+
from user_scanner.core.impersonate import impersonate_request, impersonate_validate
7+
from user_scanner.core.result import Result
8+
9+
SIGNUP_URL = "https://profile.threadless.com/artist/shops/"
10+
CSRF_RE = re.compile(r"name=['\"]csrfmiddlewaretoken['\"]\s+value=['\"]([^'\"]+)")
11+
12+
13+
def validate_threadless_shop(user: str) -> Result:
14+
shop = user.lower()
15+
show_url = f"https://{shop}.threadless.com/"
16+
if not re.fullmatch(r"[a-z0-9]{1,63}", shop):
17+
return Result.error(
18+
"Shop name must contain only lowercase letters and numbers",
19+
url=show_url,
20+
)
21+
22+
def process(page):
23+
token = CSRF_RE.search(page.text)
24+
if page.status_code != 200 or not token:
25+
return Result.error(
26+
f"Could not read Threadless signup form (HTTP {page.status_code})"
27+
)
28+
29+
response = impersonate_request(
30+
SIGNUP_URL,
31+
"POST",
32+
headers={"x-requested-with": "XMLHttpRequest"},
33+
data={
34+
"validate": "true",
35+
"shop_name": shop,
36+
"csrfmiddlewaretoken": token.group(1),
37+
},
38+
)
39+
if response.status_code != 200:
40+
return Result.error(
41+
f"Unexpected Threadless response: HTTP {response.status_code}"
42+
)
43+
44+
match response.json():
45+
case {"is_valid": True}:
46+
return Result.available()
47+
case {"is_valid": False}:
48+
profile = _shop_profile(shop)
49+
if profile:
50+
extra, media = profile
51+
return Result.taken(extra=extra, media=media)
52+
return Result.error(
53+
"Shop name is unavailable but no public shop was found"
54+
)
55+
case _:
56+
return Result.error("Unexpected Threadless response")
57+
58+
return impersonate_validate(
59+
SIGNUP_URL,
60+
process,
61+
show_url=show_url,
62+
allow_redirects=True,
63+
)
64+
65+
66+
def _shop_profile(shop: str) -> tuple[dict, dict] | None:
67+
try:
68+
response = impersonate_request(
69+
f"https://{shop}.threadless.com/", allow_redirects=True
70+
)
71+
except RequestException:
72+
return None
73+
74+
if response.status_code != 200:
75+
return None
76+
77+
document = response.text
78+
brand = _text(r'data-artist-brand="([^"]+)"', document)
79+
if not brand or brand.casefold() != shop.casefold():
80+
return None
81+
82+
about = ""
83+
try:
84+
about_response = impersonate_request(
85+
f"https://{shop}.threadless.com/about", allow_redirects=True
86+
)
87+
if about_response.status_code == 200:
88+
about_brand = _text(r'data-artist-brand="([^"]+)"', about_response.text)
89+
if about_brand and about_brand.casefold() == shop.casefold():
90+
about = about_response.text
91+
except RequestException:
92+
pass
93+
94+
social_links = []
95+
website = None
96+
for url, kind in re.findall(
97+
r'<a class="aboutSocial-cta" href="([^"]+)".*?'
98+
r'<em class="aboutSocial-title">\s*([^<]+)',
99+
about,
100+
re.DOTALL,
101+
):
102+
url, kind = html.unescape(url), kind.strip().lower()
103+
if kind == "website":
104+
website = url
105+
else:
106+
social_links.append(f"{kind}: {url}")
107+
108+
title = _text(r'<meta property="og:title"\s+content="([^"]+)"', document)
109+
name = title.split(" | ", 1)[0].removesuffix("'s Artist Shop") if title else None
110+
return (
111+
{
112+
"name": name,
113+
"description": _text(
114+
r'<meta property="og:description"\s+content="([^"]+)"',
115+
document,
116+
),
117+
"owner_id": _number(r'data-to-follow="(\d+)"', document)
118+
or _number(r'data-shop-user-id="(\d+)"', document),
119+
"owner_name": _text(
120+
r'<strong class="aboutProfile-title">(.*?)</strong>', about
121+
),
122+
"location": _text(r'<p class="aboutProfile-subtitle">(.*?)</p>', about),
123+
"headline": _clean(
124+
r'<h2 class="(?=[^"]*\baboutTitle\b)'
125+
r'(?![^"]*\b_is-hidden\b)[^"]*">(.*?)</h2>',
126+
about,
127+
),
128+
"biography": _clean(r'<div class="aboutContent-bio">(.*?)</div>', about),
129+
"website": website,
130+
"social_links": ", ".join(social_links),
131+
"search_enabled": _has_link("search", document),
132+
"gift_cards_enabled": _has_link("gift-cards", document),
133+
"wholesale_enabled": _has_link("wholesale", document),
134+
"self_reported_shop_profiles": ", ".join(
135+
_clean_text(badge)
136+
for badge in re.findall(r'<div class="badge-title">\s*([^<]+)', about)
137+
),
138+
},
139+
{
140+
"logo": _text(r'<img src="([^"]+)" alt="A logo image for', document),
141+
"cover": _text(
142+
r'<link rel="preload" media="\(min-width: 651px\)" href="([^"]+)"',
143+
document,
144+
),
145+
"preview": _text(
146+
r'<meta property="og:image"\s+content="([^"]+)"', document
147+
),
148+
"bio_photo": _text(
149+
r'<img class="aboutProfile aboutProfile--img" src="([^"]+)"',
150+
about,
151+
),
152+
},
153+
)
154+
155+
156+
def _text(pattern: str, document: str) -> str | None:
157+
match = re.search(pattern, document, re.DOTALL)
158+
return html.unescape(match.group(1)).strip() if match else None
159+
160+
161+
def _number(pattern: str, document: str) -> int | None:
162+
value = _text(pattern, document)
163+
return int(value) if value else None
164+
165+
166+
def _has_link(path: str, document: str) -> bool:
167+
return bool(
168+
re.search(
169+
rf'<a\b[^>]*\bhref="(?:https?://[^/"]+)?/{re.escape(path)}/?'
170+
r'(?:\?[^"]*)?"',
171+
document,
172+
)
173+
)
174+
175+
176+
def _clean(pattern: str, document: str) -> str | None:
177+
value = _text(pattern, document)
178+
return _clean_text(value) if value else None
179+
180+
181+
def _clean_text(value: str) -> str:
182+
return " ".join(html.unescape(re.sub(r"<[^>]+>", " ", value)).split())

0 commit comments

Comments
 (0)