I am not familiar with SourcePython.
I guess SteamWorks does have this access, but I don't
know if additional stuff is needed. You could implement it by yourself,
if you're using the URL "https://steamcommunity.com/profiles/{}/?xml=1" to
ask for profile information. This does only work, if the profile is public (I guess).
The blocking call in requests can be problematic. I used threads to solve this issue.
You must register an event, which asks Steam_API_Auth for client groups of a steam_id_64.
The callback is called as far the the class get the results. The results are cached.
During runtime the callback is called, which should give the steam_id_64 the right permission groups/flags or whatever.
This is pure python without SourcePython dependency.
I guess also requests is shipped with SourcePython. If not, you have to manage this dependency or
use urllib.request.urlopen(url).read().decode()
Code can be improved. Maybe asyncio is possible and better.
To improve it, you can change the cache_duration and maybe you work with local files on the server to cache it persistent.
Syntax: Select all
import time
from threading import Thread
import xml.etree.ElementTree as ET
import requests
from requests.exceptions import ConnectTimeout, ReadTimeout
class Steam_API_Auth:
def __init__(self, *, group_ids_64, callback, timeout=120, cache_duration=3600):
self.url = "https://steamcommunity.com/profiles/{}/?xml=1"
self.group_ids_64 = [str(gid).strip() for gid in group_ids_64]
self.callback = callback
self.cache = {}
self.timeout = timeout
self.cache_duration = cache_duration
def is_in_groups(self, steamid64):
if steamid64 in self.cache and self._timestamp_ok(steamid64):
self.callback(steamid64, self.cache[steamid64]["groups"])
return
thread = Thread(target=self._is_in_groups, args=(steamid64,), daemon=True)
thread.start()
def _timestamp_ok(self, steamid64):
timestamp = self.cache[steamid64]["timestamp"]
return time.time() - timestamp < self.cache_duration
def _is_in_groups(self, steamid64):
client_groups = [
group.get("groupID64") for group in self._get_groups(steamid64)
]
# print(client_groups)
self.cache[steamid64] = {"groups": {}, "timestamp": time.time()}
for group in self.group_ids_64:
if group in client_groups:
self.cache[steamid64]["groups"][group] = True
else:
self.cache[steamid64]["groups"][group] = False
self.callback(steamid64, self.cache[steamid64]["groups"])
def _xml_find(self, text):
xml_doc = ET.fromstring(text)
return xml_doc.findall("groups/")
def _xml_to_dict(self, group):
return {child.tag: child.text for child in group}
def _get_groups(self, steamid64):
url = self.url.format(steamid64)
exceptions = (ConnectTimeout, ReadTimeout)
try:
req = requests.get(url, timeout=self.timeout)
except exceptions:
yield {}
return
except Exception as e:
print("Exception:", type(e), "Details:", e)
if req.status_code == 200:
for group in self._xml_find(req.text):
yield self._xml_to_dict(group)
yield {}
return
def player_enters(steamid64):
"""
triggerd by an event player connects
"""
steam_group_auth.is_in_groups(steamid64)
def callback(steamid64, is_in_groups):
"""
SourcePython code to add more permissions to the client, if he is in group
"""
print("SteamID64:", steamid64, "|", "Groups:", is_in_groups)
groups = [
"103582791429524460",
"103582791429521426",
"103582791429521425",
"103582791432462448",
]
steam_group_auth = Steam_API_Auth(
group_ids_64=groups, callback=callback, cache_duration=3600
)
player_enters(76561197960915568)
# wrong
# player_enters('DeaD_EyE')
# sleep 10 seconds
# not needed later in code
time.sleep(10)